Z-IO 0.7.1.0 → 0.8.0.0

raw patch · 20 files changed

+888/−869 lines, 20 filesdep ~Z-Datadep ~timedep ~unix-timePVP ok

version bump matches the API change (PVP)

Dependency ranges changed: Z-Data, time, unix-time

API changes (from Hackage documentation)

Files

• ChangeLog.md +11/−0
• README.md +3/−4
• Z-IO.cabal +6/−7
• Z/IO/BIO.hs +359/−453
• Z/IO/BIO/Concurrent.hs +84/−28
• Z/IO/BIO/Zlib.hsc +101/−118
• Z/IO/Buffered.hs +50/−12
• Z/IO/Exception.hs +3/−3
• Z/IO/FileSystem/Base.hs +5/−5
• Z/IO/FileSystem/Threaded.hs +5/−5
• Z/IO/FileSystem/Watch.hs +7/−5
• Z/IO/FileSystem/_Shared.hs +0/−156
• Z/IO/Network/DNS.hsc +4/−4
• Z/IO/Network/SocketAddr.hsc +58/−34
• Z/IO/Resource.hs +2/−2
• cbits/hs_zlib.c +1/−2
• include/fs_shared.hs +156/−0
• test/Z/IO/BIO/ConcurrentSpec.hs +22/−21
• test/Z/IO/BIO/ZlibSpec.hs +4/−4
• test/Z/IO/BIOSpec.hs +7/−6

@@ -1,5 +1,16 @@ # Revision history for Z-IO +## 0.8.0.0  -- 2020-04-25++This is an experimental version to test new 'BIO' module.++* Rewrite `Z.IO.BIO` module, now `BIO` is push based.+* Remove `>|>`, `>~>`, `>!>`, now `BIO` nodes can be composed via funtion composition `(.)`!+* Remove `zipSource/zipBIO`, add `stepBIO/stepBIO_/runBIO_`.+* Add `zipBIO` to `Z.IO.BIO.Concurrent`, which run two BIO nodes concurrently.+* Add `ungroupingNode`, change `newGroupingNode` to use `Vector`.+* Rename `EOF` exception to `UnexpectedEOF` to avoid the clash with `EOF` pattern.+ ## 0.7.1.0  -- 2020-03-16  * Use `CPtr` from Z-Data instead of `ForeignPtr`.

@@ -5,6 +5,7 @@ [![MacOS Build Status](https://github.com/ZHaskell/z-io/workflows/osx-ci/badge.svg)](https://github.com/ZHaskell/z-io/actions) [![Windows Build Status](https://github.com/ZHaskell/z-io/workflows/win-ci/badge.svg)](https://github.com/ZHaskell/z-io/actions) [![Docker Build Status](https://github.com/ZHaskell/z-io/workflows/docker-ci/badge.svg)](https://github.com/ZHaskell/z-io/actions)+[![Gitter chat](https://badges.gitter.im/gitterHQ/gitter.svg)](https://gitter.im/Z-Haskell/community)  This package is part of [Z.Haskell](https://z.haskell.world) project, provides basic IO operations: @@ -39,10 +40,8 @@ > :{ let addr = ipv4 "13.107.21.200" 80 in withResource (initTCPClient defaultTCPClientConfig{ tcpRemoteAddr = addr}) $ \ tcp -> do-    i <- newBufferedInput tcp-    o <- newBufferedOutput tcp-    writeBuffer o "GET http://www.bing.com HTTP/1.1\r\nHost: www.bing.com\r\n\r\n"-    flushBuffer o+    (i, o) <- newBufferedIO tcp+    writeBuffer' o "GET http://www.bing.com HTTP/1.1\r\nHost: www.bing.com\r\n\r\n"     readBuffer i >>= pure . T.validate :} "HTTP/1.1 200 OK\r\nDate: Sat, 19 Sep 2020 06:11:08 GMT\r\nContent-Length: 0\r\n\r\n"

@@ -1,6 +1,6 @@ cabal-version:      2.4 name:               Z-IO-version:            0.7.1.0+version:            0.8.0.0 synopsis:           Simple and high performance IO toolkit for Haskell description:   Simple and high performance IO toolkit for Haskell, including@@ -17,6 +17,7 @@ bug-reports:        https://github.com/ZHaskell/Z-IO/issues extra-source-files:   ChangeLog.md+  include/fs_shared.hs   LICENSE   README.md   third_party/cwalk/include/cwalk.h@@ -64,7 +65,6 @@   third_party/zlib/inftrees.h   third_party/zlib/trees.h   third_party/zlib/zutil.h-  Z/IO/FileSystem/_Shared.hs  source-repository head   type:     git@@ -113,19 +113,18 @@     Z.IO.UV.FFI_Env     Z.IO.UV.Manager     Z.IO.UV.UVStream-  other-modules:-    Z.IO.UV.Win +  other-modules:      Z.IO.UV.Win   build-depends:     , base                  >=4.12  && <5.0     , containers            ^>=0.6     , exceptions            ^>=0.10     , primitive             >=0.7.1 && <0.7.2     , stm                   ^>=2.5-    , time                  >=1.9   && <=2.0-    , unix-time             >=0.4.7 && <=0.5+    , time                  >=1.9   && <2.0+    , unix-time             >=0.4.7 && <0.5     , unordered-containers  ^>=0.2-    , Z-Data                >=0.7.1 && <0.8+    , Z-Data                >=0.8.1 && <0.9    default-language:   Haskell2010   default-extensions:

@@ -1,7 +1,6 @@-{-# OPTIONS_GHC -Wno-missing-fields #-} {-| Module      : Z.IO.BIO-Description : Buffered IO interface+Description : Composable IO Loops Copyright   : (c) Dong Han, 2017-2020 License     : BSD Maintainer  : winterland1989@gmail.com@@ -37,7 +36,7 @@      withResource (initSourceFromFile origin) $ \ src ->         withResource (initSinkToFile target) $ \ sink ->-            runBIO $ src >|> base64Enc >|> zlibCompressor >|> sink+            runBIO_ $ src . base64Enc . zlibCompressor . sink  > base64AndCompressFile "test" "test.gz" -- run 'zcat "test.gz" | base64 -d' will give you original file@@ -46,27 +45,28 @@ -} module Z.IO.BIO (   -- * The BIO type-    BIO(..), Source, Sink+    BIO, pattern EOF, Source, Sink   -- ** Basic combinators-  , (>|>), (>~>), (>!>), appendSource-  , concatSource, zipSource, zipBIO+  , appendSource, concatSource, concatSource'   , joinSink, fuseSink   -- * Run BIO chain-  , runBIO-  , runSource, runSource_+  , discard+  , stepBIO, stepBIO_+  , runBIO, runBIO_   , runBlock, runBlock_, unsafeRunBlock   , runBlocks, runBlocks_, unsafeRunBlocks   -- * Make new BIO   , pureBIO, ioBIO   -- ** Source+  , initSourceFromFile+  , initSourceFromFile'   , sourceFromIO   , sourceFromList-  , initSourceFromFile   , sourceFromBuffered   , sourceTextFromBuffered   , sourceJSONFromBuffered   , sourceParserFromBuffered-  , sourceParseChunksFromBuffered+  , sourceParseChunkFromBuffered   -- ** Sink   , sinkToIO   , sinkToList@@ -74,22 +74,25 @@   , sinkToBuffered   , sinkBuilderToBuffered   -- ** Bytes specific-  , newParserNode, newReChunk, newUTF8Decoder, newMagicSplitter, newLineSplitter+  , newReChunk+  , newUTF8Decoder+  , newParserNode, newMagicSplitter, newLineSplitter   , newBase64Encoder, newBase64Decoder   , hexEncoder, newHexDecoder   -- ** Generic BIO-  , newCounterNode-  , newSeqNumNode+  , counterNode+  , seqNumNode   , newGroupingNode+  , ungroupingNode+  , consumedNode   ) where +import           Control.Concurrent.MVar import           Control.Monad import           Control.Monad.IO.Class import           Data.Bits              ((.|.)) import           Data.IORef import qualified Data.List              as List-import           Data.Sequence          (Seq (..))-import qualified Data.Sequence          as Seq import           Data.Void import           Data.Word import           System.IO.Unsafe       (unsafePerformIO)@@ -112,13 +115,13 @@  -- | A 'BIO'(blocked IO) node. ----- A 'BIO' node consist of two functions: 'push' and 'pull'. It can be used to describe different kinds of IO+-- A 'BIO' node is a push based stream transformer. It can be used to describe different kinds of IO -- devices: -- --  * @BIO inp out@ describe an IO state machine(e.g. z_stream in zlib), --    which takes some input in block, then outputs. --  * @type Source out = BIO Void out@ described an IO source, which never takes input,---    but gives output until EOF when 'pull'ed.+--    but gives output until EOF by looping. --  * @type Sink inp = BIO inp Void@ described an IO sink, which takes input and perform some IO effects, --    such as writing to terminal or files. --@@ -127,10 +130,11 @@ -- -- You can run a 'BIO' node in different ways: -----   * 'runBIO' will continuously pull value from source, push to sink until source reaches EOF.---   * 'runSource' will continuously pull value from source, and perform effects along the way.---   * 'runBlock' will supply a single block of input as whole input, and return output if there's any.---   * 'runBlocks' will supply a list of blocks as whole input, and return a list of output blocks.+--   * 'stepBIO'\/'stepBIO_' to supply a single chunk of input and step the BIO node.+--   * 'runBIO'\/'runBIO_' will supply EOF directly, which will effectively pull all values from source,+--     and push to sink until source reaches EOF.+--   * 'runBlock'\/'runBlock_' will supply a single block of input as whole input and run the BIO node.+--   * 'runBlocks'\/'runBlocks_' will supply a list of blocks as whole input and run the BIO node. -- -- Note 'BIO' usually contains some IO states, you can consider it as an opaque 'IORef': --@@ -140,232 +144,139 @@ -- 'BIO' is simply a convenient way to construct single-thread streaming computation, to use 'BIO' -- in multiple threads, check "Z.IO.BIO.Concurrent" module. ---data BIO inp out = BIO-    { push :: inp -> IO (Maybe out)-      -- ^ Push a block of input, perform some effect, and return output,-      -- if input is not enough to produce any output yet, return 'Nothing'.-    , pull :: IO (Maybe out)-      -- ^ When input reaches EOF, there may be a finalize stage to output-      -- trailing output blocks. return 'Nothing' to indicate current node-      -- reaches EOF too.-    }+type BIO inp out = (Maybe out -> IO ())     -- ^ Pass 'EOF' to indicate current node reaches EOF+                -> Maybe inp                -- ^ 'EOF' indicates upstream reaches EOF+                -> IO () +-- | Patterns for more meaningful pattern matching.+pattern EOF :: Maybe a+pattern EOF = Nothing+ -- | Type alias for 'BIO' node which never takes input. ----- 'push' is not available by type system, and 'pull' return 'Nothing' when--- reaches EOF.-type Source out = BIO Void out+-- Note when implement a 'Source', you should assume 'EOF' argument is supplied only once, and you+-- should loop to call downstream continuation with all available chunks, then write a final 'EOF'+-- to indicate EOF.+type Source x = BIO Void x  -- | Type alias for 'BIO' node which only takes input and perform effects. ----- 'push' doesn't produce any meaningful output, and 'pull' usually does a flush.-type Sink inp = BIO inp Void--instance Functor (BIO inp) where-    {-# INLINABLE fmap #-}-    fmap f BIO{..} = BIO push_ pull_-      where-        push_ inp = do-            r <- push inp-            return $! fmap f r-        pull_ = do-            r <- pull-            return $! fmap f r--infixl 3 >|>-infixl 3 >~>---- | Connect two 'BIO' nodes, feed left one's output to right one's input.-(>|>) :: HasCallStack => BIO a b -> BIO b c -> BIO a c-{-# INLINE (>|>) #-}-BIO pushA pullA >|> BIO pushB pullB = BIO push_ pull_-  where-    push_ inp = do-        x <- pushA inp-        case x of Just x' -> pushB x'-                  _       -> return Nothing-    pull_ = do-        x <- pullA-        case x of-            Just x' -> do-                y <- pushB x'-                case y of Nothing -> pull_  -- draw input from A until there's an output from B-                          _       -> return y-            _       -> pullB---- | Flipped 'fmap' for easier chaining.-(>~>) :: BIO a b -> (b -> c) -> BIO a c-{-# INLINE (>~>) #-}-(>~>) = flip fmap---- | Connect BIO to an effectful function.-(>!>) :: HasCallStack => BIO a b -> (b -> IO c) -> BIO a c-{-# INLINE (>!>) #-}-(>!>) BIO{..} f = BIO push_ pull_-  where-    push_ x = push x >>= \ r ->-        case r of Just r' -> Just <$!> f r'-                  _       -> return Nothing-    pull_ = pull >>= \ r ->-        case r of Just r' -> Just <$!> f r'-                  _       -> return Nothing+-- Note when implement a 'Sink', you should assume 'EOF' argument is supplied only once(when upstream+-- reaches EOF), you do not need to call downstream continuation before EOF, and+-- do a flush(also write a final 'EOF') when upstream reach EOF.+type Sink x = BIO x Void --- | Connect two 'BIO' source, after first reach EOF, draw element from second.-appendSource :: HasCallStack => Source a -> Source a  -> IO (Source a)+-- | Connect two 'BIO' source, after first reach EOF, draw elements from second.+appendSource :: HasCallStack => Source a -> Source a -> Source a {-# INLINE appendSource #-}-b1 `appendSource` b2 = concatSource [b1, b2]+b1 `appendSource` b2 = \ k _ ->+    b1 (\ y ->+        case y of Just _ -> k y+                  _      -> b2 k EOF) EOF  -- | Fuse two 'BIO' sinks, i.e. everything written to the fused sink will be written to left and right sink. -- -- Flush result 'BIO' will effectively flush both sink. joinSink :: HasCallStack => Sink out -> Sink out -> Sink out {-# INLINE joinSink #-}-b1 `joinSink` b2 = fuseSink [b1, b2]+b1 `joinSink` b2 = \ k mx ->+    case mx of+        Just _ -> do+            b1 discard mx+            b2 discard mx+        _ -> do+            b1 discard EOF+            b2 discard EOF+            k EOF  -- | Fuse a list of 'BIO' sinks, everything written to the fused sink will be written to every sink in the list. -- -- Flush result 'BIO' will effectively flush every sink in the list. fuseSink :: HasCallStack => [Sink out] -> Sink out {-# INLINABLE fuseSink #-}-fuseSink ss = BIO push_ pull_-  where-    push_ inp = forM_ ss (\ b -> push b inp) >> return Nothing-    pull_ = mapM_ pull ss >> return Nothing+fuseSink ss = \ k mx ->+    case mx of+        Just _ -> mapM_ (\ s -> s discard mx) ss+        _ -> do+            mapM_ (\ s -> s discard mx) ss+            k EOF  -- | Connect list of 'BIO' sources, after one reach EOF, draw element from next.-concatSource :: HasCallStack => [Source a] -> IO (Source a)+concatSource :: HasCallStack => [Source a] -> Source a {-# INLINABLE concatSource #-}-concatSource ss0 = newIORef ss0 >>= \ ref -> return (BIO{ pull = loop ref})-  where-    loop ref = do-        ss <- readIORef ref-        case ss of-            []       -> return Nothing-            (s:rest) -> do-                r <- pull s-                case r of-                    Just _ -> return r-                    _      -> writeIORef ref rest >> loop ref---- | Zip two 'BIO' source into one, reach EOF when either one reached EOF.-zipSource :: HasCallStack => Source a -> Source b -> IO (Source (a,b))-{-# INLINABLE zipSource #-}-zipSource (BIO _ pullA) (BIO _ pullB) = do-    finRef <- newIORef False-    return $ BIO { pull = do-        fin <- readIORef finRef-        if fin-        then return Nothing-        else do-            mA <- pullA-            mB <- pullB-            let r = (,) <$> mA <*> mB-            case r of-                Just _ -> return r-                _      -> writeIORef finRef True >> return Nothing-            }+concatSource = List.foldl' appendSource emptySource --- | Zip two 'BIO' node into one, reach EOF when either one reached EOF.------ The output item number should match, unmatched output will be discarded.-zipBIO :: HasCallStack => BIO a b -> BIO a c -> IO (BIO a (b, c))-{-# INLINABLE zipBIO #-}-zipBIO (BIO pushA pullA) (BIO pushB pullB) = do-    finRef <- newIORef False-    aSeqRef <- newIORef Seq.Empty-    bSeqRef <- newIORef Seq.Empty-    return (BIO (push_ aSeqRef bSeqRef) (pull_ finRef aSeqRef bSeqRef))-  where-    push_ aSeqRef bSeqRef x = do-        ma <- pushA x-        mb <- pushB x-        forM_ ma (\ a -> modifyIORef' aSeqRef (a :<|))-        forM_ mb (\ b -> modifyIORef' bSeqRef (b :<|))-        aSeq <- readIORef aSeqRef-        bSeq <- readIORef bSeqRef-        case aSeq of-            (!as :|> a) -> case bSeq of-                (!bs :|> b) -> do-                    writeIORef aSeqRef as-                    writeIORef bSeqRef bs-                    return (Just (a, b))-                _ -> return Nothing-            _ -> return Nothing+-- | A 'Source' directly write EOF to downstream.+emptySource :: Source a+{-# INLINABLE emptySource #-}+emptySource = \ k _ -> k EOF -    pull_ finRef aSeqRef bSeqRef = do-        fin <- readIORef finRef-        if fin-        then return Nothing-        else do-            aSeq <- readIORef aSeqRef-            bSeq <- readIORef bSeqRef-            ma <- case aSeq of (_ :|> a) -> return (Just a)-                               _         -> pullA-            mb <- case bSeq of (_ :|> b) -> return (Just b)-                               _         -> pullB-            case ma of-                Just a -> case mb of-                    Just b -> return (Just (a, b))-                    _      -> writeIORef finRef True >> return Nothing-                _ -> writeIORef finRef True >> return Nothing+-- | Connect list of 'BIO' sources, after one reach EOF, draw element from next.+concatSource' :: HasCallStack => Source (Source a) -> Source a+{-# INLINABLE concatSource' #-}+concatSource' ssrc = \ k _ -> ssrc (\ msrc ->+    case msrc of+        Just src -> src (\ mx ->+            case mx of Just _ -> k mx+                       _ -> return ()) EOF+        _ -> k EOF) EOF  ------------------------------------------------------------------------------- -- Run BIO --- | Run a 'BIO' loop (source >|> ... >|> sink).-runBIO :: HasCallStack => BIO Void Void -> IO ()-{-# INLINABLE runBIO #-}-runBIO BIO{..} = pull >> return ()+-- | Discards a value.+discard :: a -> IO ()+{-# INLINABLE discard #-}+discard _ = return () --- | Drain a 'BIO' source into a List in memory.-runSource :: HasCallStack => Source x -> IO [x]-{-# INLINABLE runSource #-}-runSource BIO{..} = loop pull []-  where-    loop f acc = do-        r <- f-        case r of Just r' -> loop f (r':acc)-                  _       -> return (List.reverse acc)+-- | Supply a single chunk of input to a 'BIO' and collect result.+stepBIO :: HasCallStack => BIO inp out -> inp -> IO [out]+{-# INLINABLE stepBIO #-}+stepBIO bio inp = do+    accRef <- newIORef []+    bio (mapM_  $ \ x -> modifyIORef' accRef (x:)) (Just inp)+    reverse <$> readIORef accRef --- | Drain a source without collecting result.-runSource_ :: HasCallStack => Source x -> IO ()-{-# INLINABLE runSource_ #-}-runSource_ BIO{..} = loop pull-  where-    loop f = do-        r <- f-        case r of Just _ -> loop f-                  _      -> return ()+-- | Supply a single chunk of input to a 'BIO' without collecting result.+stepBIO_ :: HasCallStack => BIO inp out -> inp -> IO ()+{-# INLINABLE stepBIO_ #-}+stepBIO_ bio = bio discard . Just --- | Supply a single block of input, then run BIO node until EOF.+-- | Run a 'BIO' loop without providing input. ----- Note many 'BIO' node will be closed or not be able to take new input after drained.+-- When used on 'Source', it starts the streaming loop.+-- When used on 'Sink', it performs a flush.+runBIO_ :: HasCallStack => BIO inp out -> IO ()+{-# INLINABLE runBIO_ #-}+runBIO_ bio = bio discard EOF++-- | Run a 'BIO' loop without providing input, and collect result.+--+-- When used on 'Source', it will collect all input chunks.+runBIO :: HasCallStack => BIO inp out -> IO [out]+{-# INLINABLE runBIO #-}+runBIO bio = do+    accRef <- newIORef []+    bio (mapM_  $ \ x -> modifyIORef' accRef (x:)) EOF+    reverse <$> readIORef accRef++-- | Run a 'BIO' loop with a single chunk of input and EOF, and collect result.+-- runBlock :: HasCallStack => BIO inp out -> inp -> IO [out] {-# INLINABLE runBlock #-}-runBlock BIO{..} inp = do-    x <- push inp-    let acc = case x of Just x' -> [x']-                        _       -> []-    loop pull acc-  where-    loop f acc = do-        r <- f-        case r of Just r' -> loop f (r':acc)-                  _       -> return (List.reverse acc)+runBlock bio inp = do+    accRef <- newIORef []+    bio (mapM_  $ \ x -> modifyIORef' accRef (x:)) (Just inp)+    bio (mapM_  $ \ x -> modifyIORef' accRef (x:)) EOF+    reverse <$> readIORef accRef --- | Supply a single block of input, then run BIO node until EOF with collecting result.+-- | Run a 'BIO' loop with a single chunk of input and EOF, without collecting result. ----- Note many 'BIO' node will be closed or not be able to take new input after drained. runBlock_ :: HasCallStack => BIO inp out -> inp -> IO () {-# INLINABLE runBlock_ #-}-runBlock_ BIO{..} inp = do-    _ <- push inp-    loop pull-  where-    loop f = do-        r <- f-        case r of Just _ -> loop f-                  _      -> return ()+runBlock_ bio inp = do+    bio discard (Just inp)+    bio discard EOF  -- | Wrap 'runBlock' into a pure interface. --@@ -375,38 +286,25 @@ {-# INLINABLE unsafeRunBlock #-} unsafeRunBlock new inp = unsafePerformIO (new >>= \ bio -> runBlock bio inp) --- | Supply blocks of input, then run BIO node until EOF.+-- | Supply blocks of input and EOF to a 'BIO', and collect results. -- -- Note many 'BIO' node will be closed or not be able to take new input after drained. runBlocks :: HasCallStack => BIO inp out -> [inp] -> IO [out] {-# INLINABLE runBlocks #-}-runBlocks BIO{..} = loop []-  where-    loop acc (inp:inps) = do-        r <- push inp-        case r of-            Just r' -> loop (r':acc) inps-            _       -> loop acc inps-    loop acc [] = loop' acc-    loop' acc = do-        r <- pull-        case r of-            Just r' -> loop' (r':acc)-            _       -> return (List.reverse acc)+runBlocks bio inps = do+    accRef <- newIORef []+    forM_ inps $ bio (mapM_  $ \ x -> modifyIORef' accRef (x:)) . Just+    bio (mapM_  $ \ x -> modifyIORef' accRef (x:)) EOF+    reverse <$> readIORef accRef --- | Supply blocks of input, then run BIO node until EOF with collecting result.+-- | Supply blocks of input and EOF to a 'BIO', without collecting results. -- -- Note many 'BIO' node will be closed or not be able to take new input after drained. runBlocks_ :: HasCallStack => BIO inp out -> [inp] -> IO () {-# INLINABLE runBlocks_ #-}-runBlocks_ bio (inp:inps) = push bio inp >> runBlocks_ bio inps-runBlocks_ bio [] = loop-  where-    loop = do-        r <- pull bio-        case r of-            Just _ -> loop-            _      -> return ()+runBlocks_ bio inps = do+    forM_ inps $ bio discard . Just+    bio discard EOF  -- | Wrap 'runBlocks' into a pure interface. --@@ -418,66 +316,72 @@ ------------------------------------------------------------------------------- -- Source --- | Source a list from memory.+-- | Source a list(or any 'Foldable') from memory. ---sourceFromList :: [a] -> IO (Source a)-sourceFromList xs0 = do-    xsRef <- newIORef xs0-    return BIO{ pull = popper xsRef }-  where-    popper xsRef = do-        xs <- readIORef xsRef-        case xs of-            (x:xs') -> do-                writeIORef xsRef xs'-                return (Just x)-            _ -> return Nothing+sourceFromList :: Foldable f => f a -> Source a+sourceFromList xs0 = \ k _ -> do+    mapM_ (k . Just) xs0+    k EOF --- | Turn a 'BufferedInput' into 'BIO' source, map EOF to Nothing.+-- | Turn a 'BufferedInput' into 'BIO' source, map EOF to EOF. -- sourceFromBuffered :: HasCallStack => BufferedInput -> Source V.Bytes {-# INLINABLE sourceFromBuffered #-}-sourceFromBuffered i = BIO{ pull = do-    readBuffer i >>= \ x -> if V.null x then return Nothing-                                        else return (Just x)}+sourceFromBuffered i = \ k _ ->+    let loop = readBuffer i >>= \ x ->+            if V.null x then k EOF else k (Just x) >> loop+    in loop  -- | Turn a `IO` action into 'Source' sourceFromIO :: HasCallStack => IO (Maybe a) -> Source a {-# INLINABLE sourceFromIO #-}-sourceFromIO io = BIO{ pull = io }+sourceFromIO io = \ k _ ->+    let loop = io >>= \ x ->+            case x of+                Just _ -> k x >> loop+                _      -> k EOF+    in loop --- | Turn a UTF8 encoded 'BufferedInput' into 'BIO' source, map EOF to Nothing.+-- | Turn a UTF8 encoded 'BufferedInput' into 'BIO' source, map EOF to EOF. -- sourceTextFromBuffered :: HasCallStack => BufferedInput -> Source T.Text {-# INLINABLE sourceTextFromBuffered #-}-sourceTextFromBuffered i = BIO{ pull = do-    readBufferText i >>= \ x -> if T.null x then return Nothing-                                            else return (Just x)}+sourceTextFromBuffered i = \ k _ ->+    let loop = readBufferText i >>= \ x ->+            if T.null x then k EOF else k (Just x) >> loop+    in loop  -- | Turn a 'JSON' encoded 'BufferedInput' into 'BIO' source, ignoring any--- whitespaces bewteen JSON objects. If EOF reached, then return Nothing.+-- whitespaces bewteen JSON objects. If EOF reached, then return 'EOF'. -- Throw 'OtherError' with name "EJSON" if JSON value is not parsed or converted.+-- sourceJSONFromBuffered :: forall a. (JSON.JSON a, HasCallStack) => BufferedInput -> Source a {-# INLINABLE sourceJSONFromBuffered #-}-sourceJSONFromBuffered = sourceParseChunksFromBuffered JSON.decodeChunks+sourceJSONFromBuffered = sourceParseChunkFromBuffered JSON.decodeChunk  -- | Turn buffered input device into a packet source, throw 'OtherError' with name @EPARSE@ if parsing fail. sourceParserFromBuffered :: HasCallStack => P.Parser a -> BufferedInput -> Source a {-# INLINABLE sourceParserFromBuffered #-}-sourceParserFromBuffered p = sourceParseChunksFromBuffered (P.parseChunks p)+sourceParserFromBuffered p = sourceParseChunkFromBuffered (P.parseChunk p)  -- | Turn buffered input device into a packet source, throw 'OtherError' with name @EPARSE@ if parsing fail.-sourceParseChunksFromBuffered :: (HasCallStack, T.Print e) => P.ParseChunks IO V.Bytes e a -> BufferedInput -> Source a-{-# INLINABLE sourceParseChunksFromBuffered #-}-sourceParseChunksFromBuffered cp bi = BIO{ pull = do-    bs <- readBuffer bi-    if V.null bs-       then return Nothing-       else do-           (rest, r) <- cp (readBuffer bi) bs-           unReadBuffer rest bi-           case r of Right v -> return (Just v)-                     Left e  -> throwOtherError "EPARSE" (T.toText e) }+sourceParseChunkFromBuffered :: (HasCallStack, T.Print e)+                              => (V.Bytes -> P.Result e a) -> BufferedInput -> Source a+{-# INLINABLE sourceParseChunkFromBuffered #-}+sourceParseChunkFromBuffered pc bi = \ k _ ->+    let loopA = do+            bs <- readBuffer bi+            if V.null bs+            then k EOF+            else loopB bs+        loopB bs = do+            (rest, r) <- P.parseChunks pc (readBuffer bi) bs+            case r of Right v -> k (Just v)+                      Left e  -> throwOtherError "EPARSE" (T.toText e)+            if V.null rest+            then loopA+            else loopB rest+    in loopA  -- | Turn a file into a 'V.Bytes' source. initSourceFromFile :: HasCallStack => CBytes -> Resource (Source V.Bytes)@@ -486,25 +390,32 @@     f <- FS.initFile p FS.O_RDONLY FS.DEFAULT_FILE_MODE     liftIO (sourceFromBuffered <$> newBufferedInput f) +-- | Turn a file into a 'V.Bytes' source with given chunk size.+initSourceFromFile' :: HasCallStack => CBytes -> Int -> Resource (Source V.Bytes)+{-# INLINABLE initSourceFromFile' #-}+initSourceFromFile' p bufSiz = do+    f <- FS.initFile p FS.O_RDONLY FS.DEFAULT_FILE_MODE+    liftIO (sourceFromBuffered <$> newBufferedInput' bufSiz f)+ -------------------------------------------------------------------------------- -- Sink  -- | Turn a 'BufferedOutput' into a 'V.Bytes' sink. sinkToBuffered :: HasCallStack => BufferedOutput -> Sink V.Bytes {-# INLINABLE sinkToBuffered #-}-sinkToBuffered bo = BIO push_ pull_-  where-    push_ inp = writeBuffer bo inp >> pure Nothing-    pull_ = flushBuffer bo >> pure Nothing+sinkToBuffered bo = \ k mbs ->+    case mbs of+        Just bs -> writeBuffer bo bs+        _       -> flushBuffer bo >> k EOF  -- | Turn a 'BufferedOutput' into a 'B.Builder' sink. -- sinkBuilderToBuffered :: HasCallStack => BufferedOutput -> Sink (B.Builder a) {-# INLINABLE sinkBuilderToBuffered #-}-sinkBuilderToBuffered bo = BIO push_ pull_-  where-    push_ inp = writeBuilder bo inp >> pure Nothing-    pull_ = flushBuffer bo >> pure Nothing+sinkBuilderToBuffered bo = \ k mbs ->+    case mbs of+        Just bs -> writeBuilder bo bs+        _       -> flushBuffer bo >> k EOF  -- | Turn a file into a 'V.Bytes' sink. --@@ -518,25 +429,33 @@  -- | Turn an `IO` action into 'BIO' sink. ----- 'push' will call `IO` action with input chunk, `pull` has no effect. sinkToIO :: HasCallStack => (a -> IO ()) -> Sink a {-# INLINABLE sinkToIO #-}-sinkToIO f = BIO push_ pull_-  where-    push_ x = f x >> pure Nothing-    pull_ = pure Nothing+sinkToIO f = \ k ma ->+    case ma of+        Just a -> f a+        _ -> k EOF --- | Sink to a list in memory.+-- | Turn an `IO` action(and a flush action), into 'BIO' sink. ----- The list's 'IORef' is not thread safe here,--- and list items are in reversed order during sinking(will be reversed when flushed, i.e. pulled),--- Please don't use it in multiple thread.+sinkToIO' :: HasCallStack => (a -> IO ()) -> IO () -> Sink a+{-# INLINABLE sinkToIO' #-}+sinkToIO' f flush = \ k ma ->+    case ma of+        Just a -> f a+        _ -> flush >> k EOF++-- | Sink to a list in memory. ---sinkToList :: IO (IORef [a], Sink a)+-- The 'MVar' will be empty during sinking, and will be filled after sink receives an EOF.+sinkToList :: IO (MVar [a], Sink a) sinkToList = do     xsRef <- newIORef []-    return (xsRef, BIO (\ x -> modifyIORef xsRef (x:) >> return Nothing)-                       (modifyIORef xsRef reverse >> return Nothing))+    rRef <- newEmptyMVar+    return (rRef, sinkToIO' (\ x -> modifyIORef xsRef (x:))+                            (do modifyIORef xsRef reverse+                                xs <- readIORef xsRef+                                putMVar rRef xs))  -------------------------------------------------------------------------------- -- Nodes@@ -545,14 +464,18 @@ -- -- BIO node made with this funtion are stateless, thus can be reused across chains. pureBIO :: (a -> b) -> BIO a b-pureBIO f = BIO (\ x -> let !r = f x in return (Just r)) (return Nothing)+{-# INLINE pureBIO #-}+pureBIO f = \ k x -> k (f <$> x)  -- | BIO node from an IO function. -- -- BIO node made with this funtion may not be stateless, it depends on if the IO function use -- IO state.-ioBIO :: (HasCallStack => a -> IO b) -> BIO a b-ioBIO f = BIO (\ x -> Just <$!> f x) (return Nothing)+ioBIO :: HasCallStack => (a -> IO b) -> BIO a b+{-# INLINE ioBIO #-}+ioBIO f = \ k x ->+    case x of Just x' -> f x' >>= k . Just+              _ -> k EOF  -- | Make a chunk size divider. --@@ -563,75 +486,55 @@ {-# INLINABLE newReChunk #-} newReChunk n = do     trailingRef <- newIORef V.empty-    return (BIO (push_ trailingRef) (pull_ trailingRef))-  where-    push_ trailingRef bs = do-        trailing <- readIORef trailingRef-        let chunk =  trailing `V.append` bs-            l = V.length chunk-        if l >= n-        then do-            let l' = l - (l `rem` n)-                (chunk', rest) = V.splitAt l' chunk-            writeIORef trailingRef rest-            return (Just chunk')-        else do-            writeIORef trailingRef chunk-            return Nothing-    pull_ trailingRef = do-        trailing <- readIORef trailingRef-        if V.null trailing-        then return Nothing-        else do-            writeIORef trailingRef V.empty-            return (Just trailing)+    return $ \ k mbs ->+        case mbs of+            Just bs -> do+                trailing <- readIORef trailingRef+                let chunk =  trailing `V.append` bs+                    l = V.length chunk+                if l >= n+                then do+                    let l' = l - (l `rem` n)+                        (chunk', rest) = V.splitAt l' chunk+                    writeIORef trailingRef rest+                    k (Just chunk')+                else writeIORef trailingRef chunk+            _ -> do+                trailing <- readIORef trailingRef+                unless (V.null trailing) $ do+                    writeIORef trailingRef V.empty+                    k (Just trailing)+                k EOF  -- | Read buffer and parse with 'Parser'. ----- This function will continuously draw data from input before parsing finish.--- Unconsumed bytes will be returned to buffer.+-- This function will turn a 'Parser' into a 'BIO', throw 'OtherError' with name @EPARSE@ if parsing fail. ----- Return 'Nothing' if reach EOF before parsing, throw 'OtherError' with name @EPARSE@ if parsing fail. newParserNode :: HasCallStack => P.Parser a -> IO (BIO V.Bytes a) {-# INLINABLE newParserNode #-} newParserNode p = do-    -- type LastParseState = Either V.Bytes (V.Bytes -> P.Result)-    resultRef <- newIORef (Left V.empty)-    return (BIO (push_ resultRef) (pull_ resultRef))-  where-    push_ resultRef bs = do-        lastResult <- readIORef resultRef-        let (chunk, f) = case lastResult of-                Left trailing -> (trailing `V.append` bs, P.parseChunk p)-                Right x       -> (bs, x)-        case f chunk of-            P.Success a trailing' -> do-                writeIORef resultRef (Left trailing')-                return (Just a)-            P.Failure e _ ->-                throwOtherError "EPARSE" (T.toText e)-            P.Partial f' -> do-                writeIORef resultRef (Right f')-                return Nothing--    pull_ resultRef = do+    -- type LastParseState = Maybe (V.Bytes -> P.Result)+    resultRef <- newIORef EOF+    return $ \ k mbs -> do+        let loop f chunk = case f chunk of+                P.Success a trailing -> do+                    k (Just a)+                    unless (V.null trailing) (loop f trailing)+                P.Partial f' ->+                    writeIORef resultRef (Just f')+                P.Failure e _ ->+                    throwOtherError "EPARSE" (T.toText e)         lastResult <- readIORef resultRef-        case lastResult of-            Left trailing ->-                if V.null trailing-                then return Nothing-                else lastChunk resultRef (P.parseChunk p) trailing-            Right f -> lastChunk resultRef f V.empty--    lastChunk resultRef f chunk =-        case f chunk of-            P.Success a trailing' -> do-                writeIORef resultRef (Left trailing')-                return (Just a)-            P.Failure e _ ->-                throwOtherError "EPARSE" (T.toText e)-            P.Partial _ ->-                throwOtherError "EPARSE" "last chunk partial parse"+        case mbs of+            Just bs -> do+                let f = case lastResult of+                        Just x    -> x+                        _         -> P.parseChunk p+                loop f bs+            _ ->+                case lastResult of+                    Just f -> loop f V.empty+                    _ -> k EOF  -- | Make a new UTF8 decoder, which decode bytes streams into text streams. --@@ -647,34 +550,32 @@ {-# INLINABLE newUTF8Decoder #-} newUTF8Decoder = do     trailingRef <- newIORef V.empty-    return (BIO (push_ trailingRef) (pull_ trailingRef))-  where-    push_ trailingRef bs = do-        trailing <- readIORef trailingRef-        let chunk =  trailing `V.append` bs-            (V.PrimVector arr s l) = chunk-        if l > 0 && T.decodeCharLen arr s <= l-        then do-            let (i, _) = V.findR (\ w -> w >= 0b11000000 || w <= 0b01111111) chunk-            if (i == -1)-            then throwOtherError "EINVALIDUTF8" "invalid UTF8 bytes"-            else do-                if T.decodeCharLen arr (s + i) > l - i+    return $ \ k mbs -> do+        case mbs of+            Just bs -> do+                trailing <- readIORef trailingRef+                let chunk =  trailing `V.append` bs+                    (V.PrimVector arr s l) = chunk+                if l > 0 && T.decodeCharLen arr s <= l                 then do-                    writeIORef trailingRef (V.fromArr arr (s+i) (l-i))-                    return (Just (T.validate (V.fromArr arr s i)))-                else do-                    writeIORef trailingRef V.empty-                    return (Just (T.validate chunk))-        else do-            writeIORef trailingRef chunk-            return Nothing+                    let (i, _) = V.findR (\ w -> w >= 0b11000000 || w <= 0b01111111) chunk+                    if (i == -1)+                    then throwOtherError "EINVALIDUTF8" "invalid UTF8 bytes"+                    else do+                        if T.decodeCharLen arr (s + i) > l - i+                        then do+                            writeIORef trailingRef (V.fromArr arr (s+i) (l-i))+                            k (Just (T.validate (V.fromArr arr s i)))+                        else do+                            writeIORef trailingRef V.empty+                            k (Just (T.validate chunk))+                else writeIORef trailingRef chunk -    pull_ trailingRef = do-        trailing <- readIORef trailingRef-        if V.null trailing-        then return Nothing-        else throwOtherError "EINVALIDUTF8" "invalid UTF8 bytes"+            _ -> do+                trailing <- readIORef trailingRef+                if V.null trailing+                then k EOF+                else throwOtherError "EINVALIDUTF8" "invalid UTF8 bytes"  -- | Make a new stream splitter based on magic byte. --@@ -682,32 +583,24 @@ {-# INLINABLE newMagicSplitter #-} newMagicSplitter magic = do     trailingRef <- newIORef V.empty-    return (BIO (push_ trailingRef) (pull_ trailingRef))-  where-    push_ trailingRef bs = do-        trailing <- readIORef trailingRef-        let chunk =  trailing `V.append` bs-        case V.elemIndex magic chunk of-            Just i -> do-                let (line, rest) = V.splitAt (i+1) chunk-                writeIORef trailingRef rest-                return (Just line)-            _ -> do-                writeIORef trailingRef chunk-                return Nothing--    pull_ trailingRef = do-        chunk <- readIORef trailingRef-        if V.null chunk-        then return Nothing-        else case V.elemIndex magic chunk of-            Just i -> do-                let (line, rest) = V.splitAt (i+1) chunk-                writeIORef trailingRef rest-                return (Just line)+    return $ \ k mx ->+        case mx of+            Just bs -> do+                trailing <- readIORef trailingRef+                let loop chunk = case V.elemIndex magic chunk of+                        Just i -> do+                            -- TODO: looping+                            let (line, rest) = V.splitAt (i+1) chunk+                            k (Just line)+                            loop rest+                        _ -> writeIORef trailingRef chunk+                loop (trailing `V.append` bs)             _ -> do-                writeIORef trailingRef V.empty-                return (Just chunk)+                chunk <- readIORef trailingRef+                unless (V.null chunk) $ do+                    writeIORef trailingRef V.empty+                    k (Just chunk)+                k EOF  -- | Make a new stream splitter based on linefeed(@\r\n@ or @\n@). --@@ -716,7 +609,7 @@ {-# INLINABLE newLineSplitter #-} newLineSplitter = do     s <- newMagicSplitter 10-    return (s >~> dropLineEnd)+    return (s . pureBIO dropLineEnd)   where     dropLineEnd bs@(V.PrimVector arr s l) =         case bs `V.indexMaybe` (l-2) of@@ -730,14 +623,14 @@ {-# INLINABLE newBase64Encoder #-} newBase64Encoder = do     re <- newReChunk 3-    return (re >~> base64Encode)+    return (re . pureBIO base64Encode)  -- | Make a new base64 decoder node. newBase64Decoder :: HasCallStack => IO (BIO V.Bytes V.Bytes) {-# INLINABLE newBase64Decoder #-} newBase64Decoder = do     re <- newReChunk 4-    return (re >~> base64Decode')+    return (re . pureBIO base64Decode')  -- | Make a hex encoder node. --@@ -752,69 +645,82 @@ {-# INLINABLE newHexDecoder #-} newHexDecoder = do     re <- newReChunk 2-    return (re >~> hexDecode')+    return (re . pureBIO hexDecode')  -- | Make a new BIO node which counts items flow throught it. ----- Returned 'Counter' is increased atomically, it's safe to read \/ reset the counter from other threads.-newCounterNode :: IO (Counter, BIO a a)-{-# INLINABLE newCounterNode #-}-newCounterNode = do-    c <- newCounter 0-    return (c, BIO (push_ c) (return Nothing))+-- 'Counter' is increased atomically, it's safe to read \/ reset the counter from other threads.+counterNode :: Counter -> BIO a a+{-# INLINABLE counterNode #-}+counterNode c = ioBIO inc   where-    push_ c x = do+    inc x = do         atomicAddCounter_ c 1-        return (Just x)+        return x  -- | Make a new BIO node which counts items, and label item with a sequence number. ----- Returned 'Counter' is increased atomically, it's safe to read \/ reset the counter from other threads.-newSeqNumNode :: IO (Counter, BIO a (Int, a))-{-# INLINABLE newSeqNumNode #-}-newSeqNumNode = do-    c <- newCounter 0-    return (c, BIO (push_ c) (return Nothing))+-- 'Counter' is increased atomically, it's safe to read \/ reset the counter from other threads.+seqNumNode :: Counter -> BIO a (Int, a)+{-# INLINABLE seqNumNode #-}+seqNumNode c = ioBIO inc   where-    push_ c x = do-        !i <- atomicAddCounter c 1-        return (Just (i, x))+    inc x = do+        i <- atomicAddCounter c 1+        return (i, x)  -- | Make a BIO node grouping items into fixed size arrays. ---newGroupingNode :: Int -> IO (BIO a (A.SmallArray a))+-- Trailing items are directly returned.+newGroupingNode :: Int -> IO (BIO a (V.Vector a)) {-# INLINABLE newGroupingNode #-} newGroupingNode n     | n < 1 =  newGroupingNode 1     | otherwise = do         c <- newCounter 0         arrRef <- newIORef =<< A.newArr n-        return (BIO (push_ c arrRef) (pull_ c arrRef))-  where-    push_ c arrRef x = do-        i <- readPrimIORef c-        if i == n - 1-        then do-            marr <- readIORef arrRef-            A.writeArr marr i x-            writePrimIORef c 0-            writeIORef arrRef =<< A.newArr n-            return . Just =<< A.unsafeFreezeArr marr-        else do-            marr <- readIORef arrRef-            A.writeArr marr i x-            writePrimIORef c (i+1)-            return Nothing-    pull_ c arrRef = do-        i <- readPrimIORef c-        if i /= 0-        then do-            marr <- readIORef arrRef-#if MIN_VERSION_base(4,14,0)-            A.shrinkMutableArr marr i-            return . Just =<< A.unsafeFreezeArr marr-#else-            marr' <- A.resizeMutableArr marr i-            return . Just =<< A.unsafeFreezeArr marr'-#endif-        else return Nothing+        return $ \ k mx ->+            case mx of+                Just x -> do+                    i <- readPrimIORef c+                    if i == n - 1+                    then do+                        marr <- readIORef arrRef+                        A.writeArr marr i x+                        writePrimIORef c 0+                        writeIORef arrRef =<< A.newArr n+                        arr <- A.unsafeFreezeArr marr+                        k . Just $! V.fromArr arr 0 n+                    else do+                        marr <- readIORef arrRef+                        A.writeArr marr i x+                        writePrimIORef c (i+1)+                _ -> do+                    i <- readPrimIORef c+                    if i /= 0+                    then do+                        writePrimIORef c 0+                        marr <- readIORef arrRef+                        A.shrinkMutableArr marr i+                        arr <- A.unsafeFreezeArr marr+                        k . Just $! V.fromArr arr 0 i+                    else k EOF++-- | A BIO node flatten items.+--+ungroupingNode :: BIO (V.Vector a) a+{-# INLINABLE ungroupingNode #-}+ungroupingNode = \ k mx ->+    case mx of+        Just x -> V.traverseVec_ (k . Just) x+        _      -> k EOF++-- | A BIO node which write 'True' to 'IORef' when 'EOF' is reached.+consumedNode :: IORef Bool -> BIO a a+{-# INLINABLE consumedNode #-}+consumedNode ref = \ k mx -> case mx of+    Just _ -> k mx+    _ -> do writeIORef ref True+            k EOF++

@@ -36,7 +36,7 @@  forkIO $ do     ...-    (runBIO $ ... >|> sink) -- producer using BIO+    (runBIO $ ... . sink) -- producer using BIO         `onException` (pull sink)  --------------------------------------------------------------------------------@@ -46,11 +46,11 @@     ...     r <- pull src           -- consumer using pull     case r of Just r' -> ...-              _ -> ...      -- Nothing indicate all producers reached EOF+              _ -> ...      -- EOF indicate all producers reached EOF  forkIO $ do     ...-    runBIO $ src >|> ...    -- consumer using BIO+    runBIO $ src . ...    -- consumer using BIO @  -}@@ -58,12 +58,51 @@ module Z.IO.BIO.Concurrent where  import Control.Monad+import Control.Concurrent import Control.Concurrent.STM+import qualified Data.Sequence as Seq+import Data.Sequence (Seq((:<|),(:|>))) import GHC.Natural import Z.IO.BIO import Z.Data.PrimRef import Z.IO.Exception +-- | Zip two BIO node by running them concurrently.+--+-- This implementation use 'MVar' to synchronize two BIO's output, which has some implications:+--+--   * Two node should output same numebr of results.+--   * If the number differs, one node maybe+--+zipBIO :: BIO a b -> BIO a c -> BIO a (b,c)+zipBIO b1 b2 = \ k mx -> do+    bEOF <- newTVarIO False+    cEOF <- newTVarIO False+    bBuf <- newTVarIO Seq.empty+    cBuf <- newTVarIO Seq.empty+    _ <- forkIO (b1 (f bBuf bEOF) mx)+    _ <- forkIO (b2 (f cBuf cEOF) mx)+    loop k bBuf cBuf bEOF cEOF+  where+    f xBuf xEOF = \ mx ->+        case mx of+            Just x -> atomically $ modifyTVar' xBuf (:|> x)+            _ -> atomically $ writeTVar xEOF True++    loop k bBuf cBuf bEOF cEOF = join . atomically $ do+        bs <- readTVar bBuf+        cs <- readTVar cBuf+        beof <- readTVar bEOF+        ceof <- readTVar cEOF+        case bs of+            b :<| bs' -> case cs of+                c :<| cs' -> do+                    writeTVar bBuf bs'+                    writeTVar cBuf cs'+                    return (k (Just (b, c)) >> loop k bBuf cBuf bEOF cEOF)+                _ -> if ceof then return (k EOF) else retry+            _ -> if beof then return (k EOF) else retry+ -- | Make an unbounded queue and a pair of sink and souce connected to it. newTQueueNode :: Int -- ^ number of producers               -> IO (Sink a, Source a)@@ -71,15 +110,21 @@     q <- newTQueueIO     ec <- newCounter 0     return-        ( BIO (\ x -> atomically (writeTQueue q (Just x)) >> return Nothing)-                (do i <- atomicAddCounter' ec 1-                    when (i == n) (atomically (writeTQueue q Nothing))-                    return Nothing)-        , BIO { pull = uninterruptibleMask $ \ restore -> do+        ( \ k mx -> case mx of+                Just _ -> atomically (writeTQueue q mx)+                _ -> do+                    i <- atomicAddCounter' ec 1+                    when (i == n) $ do+                        atomically (writeTQueue q EOF)+                        k EOF++        , \ k _ ->+            let loop = uninterruptibleMask $ \ restore -> do                     x <- restore $ atomically (readTQueue q)-                    case x of Just _ -> return x-                              _ -> do atomically (unGetTQueue q Nothing)-                                      return Nothing})+                    case x of Just _ -> k x >> loop+                              _ -> do atomically (unGetTQueue q EOF)+                                      k EOF+            in loop)  -- | Make an bounded queue and a pair of sink and souce connected to it. newTBQueueNode :: Int       -- ^ number of producers@@ -89,15 +134,21 @@     q <- newTBQueueIO bound     ec <- newCounter 0     return-        ( BIO (\ x -> atomically (writeTBQueue q (Just x)) >> return Nothing)-                (do i <- atomicAddCounter' ec 1-                    when (i == n) (atomically (writeTBQueue q Nothing))-                    return Nothing)-        , BIO { pull = uninterruptibleMask $ \ restore -> do+        ( \ k mx -> case mx of+                Just _ -> atomically (writeTBQueue q mx)+                _ -> do+                    i <- atomicAddCounter' ec 1+                    when (i == n) $ do+                        atomically (writeTBQueue q EOF)+                        k EOF++        , \ k _ ->+            let loop = uninterruptibleMask $ \ restore -> do                     x <- restore $ atomically (readTBQueue q)-                    case x of Just _ -> return x-                              _      -> do atomically (unGetTBQueue q Nothing)-                                           return Nothing})+                    case x of Just _ -> k x >> loop+                              _ -> do atomically (unGetTBQueue q EOF)+                                      k EOF+            in loop)  -- | Make a broadcast chan and a sink connected to it, and a function return sources to receive broadcast message. newBroadcastTChanNode :: Int                        -- ^ number of producers@@ -107,12 +158,17 @@     ec <- newCounter 0     let dupSrc = do             c <- atomically (dupTChan b)-            return (BIO { pull = do-                            x <- atomically (readTChan c)-                            case x of Just _ -> return x-                                      _ -> return Nothing })-    return ( BIO (\ x -> atomically (writeTChan b (Just x)) >> return Nothing)-                    (do i <- atomicAddCounter' ec 1-                        when (i == n) (atomically (writeTChan b Nothing))-                        return Nothing)-           , dupSrc)+            return $ \ k _ ->+                let loop = do+                        x <- atomically (readTChan c)+                        case x of Just _ -> k x >> loop+                                  _ -> k EOF+                in loop++    return+        (\ k mx -> case mx of+            Just _ -> atomically (writeTChan b mx)+            _ -> do i <- atomicAddCounter' ec 1+                    when (i == n) (atomically (writeTChan b EOF))+                    k EOF+       , dupSrc)

@@ -12,7 +12,7 @@ @ -- add compressor to your BIO chain to compress streaming blocks of 'V.Bytes'. (_, zlibCompressor) <- newCompress defaultCompressConfig{compressWindowBits = 31}-runBIO $ src >|> zlibCompressor >|> sink+runBIO $ src . zlibCompressor . sink @  -}@@ -54,7 +54,6 @@  import           Control.Monad import           Data.IORef-import qualified Data.List          as List import           Data.Word import           Foreign            hiding (void) import           Foreign.C@@ -126,9 +125,8 @@         defaultMemLevel V.empty Z_DEFAULT_STRATEGY V.defaultChunkSize  -- | A foreign pointer to a zlib\'s @z_stream_s@ struct.-data ZStream = ZStream -    {-# UNPACK #-} !(CPtr ZStream) -    {-# UNPACK #-} !(IORef Bool)+newtype ZStream = ZStream (CPtr ZStream) deriving (Eq, Ord, Show)+                                         deriving newtype Print  -- | Make a new compress node. --@@ -137,10 +135,10 @@             => CompressConfig             -> IO (ZStream, BIO V.Bytes V.Bytes) newCompress (CompressConfig level windowBits memLevel dict strategy bufSiz) = do-    (zs, _) <- newCPtrUnsafe -        (\ mba## -> do-            ps <- throwOOMIfNull (create_z_stream mba##)-            throwZlibIfMinus_ $ deflate_init2 ps level windowBits memLevel strategy)+    zs <- newCPtr'+        (do ps <- throwOOMIfNull create_z_stream+            throwZlibIfMinus_ $ deflate_init2 ps level windowBits memLevel strategy+            return ps)         free_z_stream_deflate      unless (V.null dict) .  withCPtr zs $ \ ps -> do@@ -148,60 +146,51 @@             deflate_set_dictionary ps pdict off (fromIntegral $ len)      buf <- A.newPinnedPrimArray bufSiz-    set_avail_out zs buf bufSiz     bufRef <- newIORef buf-    finRef <- newIORef False-    return (ZStream zs finRef, BIO (zwrite zs bufRef) (zflush finRef zs bufRef []))-  where-    zwrite zs bufRef input = do-        set_avail_in zs input (V.length input)-        zloop zs bufRef []+    set_avail_out zs buf bufSiz -    zloop zs bufRef acc = do-        oavail :: CUInt <- withCPtr zs $ \ ps -> do-            throwZlibIfMinus_ (deflate ps (#const Z_NO_FLUSH))-            (#peek struct z_stream_s, avail_out) ps-        if oavail == 0-        then do-            oarr <- A.unsafeFreezeArr =<< readIORef bufRef+    let newOutBuffer = do             buf' <- A.newPinnedPrimArray bufSiz-            set_avail_out zs buf' bufSiz             writeIORef bufRef buf'-            zloop zs bufRef (V.PrimVector oarr 0 bufSiz : acc)-        else do-            let output = V.concat (List.reverse acc)-            if V.null output then return Nothing-                             else return (Just output)+            set_avail_out zs buf' bufSiz -    zflush finRef zs bufRef acc = do-        fin <- readIORef finRef-        if fin-        then return Nothing-        else do-            buf <- readIORef bufRef-            (r, osiz) <- withCPtr zs $ \ ps -> do-                r <- throwZlibIfMinus (deflate ps (#const Z_FINISH))-                oavail :: CUInt <- (#peek struct z_stream_s, avail_out) ps-                return (r, bufSiz - fromIntegral oavail)-            if (r /= (#const Z_STREAM_END) && osiz /= 0)-            then do-                oarr <- A.unsafeFreezeArr buf-                buf' <- A.newPinnedPrimArray bufSiz-                set_avail_out zs buf' bufSiz-                writeIORef bufRef buf'-                zflush finRef zs bufRef (V.PrimVector oarr 0 osiz : acc)-            else do-                oarr <- A.unsafeFreezeArr buf-                let trailing = V.concat . List.reverse $ V.PrimVector oarr 0 osiz : acc-                -- stream ends-                writeIORef finRef True-                if V.null trailing then return Nothing else return (Just trailing)+    return (ZStream zs, \ k mbs -> case mbs of+        Just bs -> do+            set_avail_in zs bs (V.length bs)+            let loop = do+                    oavail :: CUInt <- withCPtr zs $ \ ps -> do+                        throwZlibIfMinus_ (deflate ps (#const Z_NO_FLUSH))+                        (#peek struct z_stream_s, avail_out) ps+                    when (oavail == 0) $ do+                        oarr <- A.unsafeFreezeArr =<< readIORef bufRef+                        k (Just (V.PrimVector oarr 0 bufSiz))+                        newOutBuffer           +                        loop+            loop+        _ -> +            let loop = do+                    (r, osiz) <- withCPtr zs $ \ ps -> do+                        r <- throwZlibIfMinus (deflate ps (#const Z_FINISH))+                        oavail :: CUInt <- (#peek struct z_stream_s, avail_out) ps+                        return (r, bufSiz - fromIntegral oavail)+                    if (r /= (#const Z_STREAM_END) && osiz /= 0)+                    then do+                        oarr <- A.unsafeFreezeArr =<< readIORef bufRef+                        k (Just (V.PrimVector oarr 0 osiz))+                        newOutBuffer+                        loop+                    else do+                        -- stream ends+                        when (osiz /= 0) $ do+                            oarr <- A.unsafeFreezeArr =<< readIORef bufRef+                            k (Just (V.PrimVector oarr 0 osiz))+                        k EOF+            in loop)  -- | Reset compressor's state so that related 'BIO' can be reused. compressReset :: ZStream -> IO ()-compressReset (ZStream fp finRef) = do+compressReset (ZStream fp) = do     throwZlibIfMinus_ (withCPtr fp deflateReset)-    writeIORef finRef False  -- | Compress some bytes. compress :: HasCallStack => CompressConfig -> V.Bytes -> V.Bytes@@ -226,82 +215,76 @@ -- The returned 'BIO' node can be reused only if you call 'decompressReset' on the 'ZStream'. newDecompress :: DecompressConfig -> IO (ZStream, BIO V.Bytes V.Bytes) newDecompress (DecompressConfig windowBits dict bufSiz) = do-    (zs, _) <- newCPtrUnsafe -        (\ mba## -> do-            ps <- throwOOMIfNull (create_z_stream mba##)-            throwZlibIfMinus_ $ inflate_init2 ps windowBits)+    zs <- newCPtr'+        (do ps <- throwOOMIfNull create_z_stream+            throwZlibIfMinus_ $ inflate_init2 ps windowBits+            return ps)         free_z_stream_inflate      buf <- A.newPinnedPrimArray bufSiz-    set_avail_out zs buf bufSiz     bufRef <- newIORef buf-    finRef <- newIORef False-    return (ZStream zs finRef, BIO (zwrite zs bufRef) (zflush finRef zs bufRef []))-  where-    zwrite zs bufRef input = do-        set_avail_in zs input (V.length input)-        zloop zs bufRef []+    set_avail_out zs buf bufSiz -    zloop zs bufRef acc = do-        oavail :: CUInt <- withCPtr zs $ \ ps -> do-            r <- throwZlibIfMinus (inflate ps (#const Z_NO_FLUSH))-            when (r == (#const Z_NEED_DICT)) $-                if V.null dict-                then throwIO (ZlibException "Z_NEED_DICT" callStack)-                else do-                    throwZlibIfMinus_ . withPrimVectorUnsafe dict $ \ pdict off len ->-                        inflate_set_dictionary ps pdict off (fromIntegral len)-                    throwZlibIfMinus_ (inflate ps (#const Z_NO_FLUSH))-            (#peek struct z_stream_s, avail_out) ps-        if oavail == 0-        then do-            oarr <- A.unsafeFreezeArr =<< readIORef bufRef+    let newOutBuffer = do             buf' <- A.newPinnedPrimArray bufSiz-            set_avail_out zs buf' bufSiz             writeIORef bufRef buf'-            zloop zs bufRef (V.PrimVector oarr 0 bufSiz : acc)-        else do-            let output = V.concat (List.reverse acc)-            if V.null output then return Nothing-                             else return (Just output)+            set_avail_out zs buf' bufSiz -    zflush finRef zs bufRef acc = do-        fin <- readIORef finRef-        if fin-        then return Nothing-        else do-            buf <- readIORef bufRef-            (r, osiz) <- withCPtr zs $ \ ps -> do-                r <- throwZlibIfMinus (inflate ps (#const Z_FINISH))-                r' <- if r == (#const Z_NEED_DICT)-                then if V.null dict-                    then throwIO (ZlibException "Z_NEED_DICT" callStack)+    return (ZStream zs, \ k mbs -> case mbs of+        Just bs -> do+            set_avail_in zs bs (V.length bs)++            let loop = do+                    oavail :: CUInt <- withCPtr zs $ \ ps -> do+                        r <- throwZlibIfMinus (inflate ps (#const Z_NO_FLUSH))+                        when (r == (#const Z_NEED_DICT)) $+                            if V.null dict+                            then throwIO (ZlibException "Z_NEED_DICT" callStack)+                            else do+                                throwZlibIfMinus_ . withPrimVectorUnsafe dict $ \ pdict off len ->+                                    inflate_set_dictionary ps pdict off (fromIntegral len)+                                throwZlibIfMinus_ (inflate ps (#const Z_NO_FLUSH))+                        (#peek struct z_stream_s, avail_out) ps++                    when (oavail == 0) $ do+                        oarr <- A.unsafeFreezeArr =<< readIORef bufRef+                        k (Just (V.PrimVector oarr 0 bufSiz))+                        newOutBuffer+                        loop+            loop++        _ -> +            let loop = do+                    (r, osiz) <- withCPtr zs $ \ ps -> do+                        r <- throwZlibIfMinus (inflate ps (#const Z_FINISH))+                        r' <- if r == (#const Z_NEED_DICT)+                        then if V.null dict+                            then throwIO (ZlibException "Z_NEED_DICT" callStack)+                            else do+                                throwZlibIfMinus_ . withPrimVectorUnsafe dict $ \ pdict off len ->+                                    inflate_set_dictionary ps pdict off (fromIntegral len)+                                throwZlibIfMinus (inflate ps (#const Z_FINISH))+                        else return r+                        oavail :: CUInt <- (#peek struct z_stream_s, avail_out) ps+                        return (r', bufSiz - fromIntegral oavail)+                    if (r /= (#const Z_STREAM_END) && osiz /= 0)+                    then do+                        oarr <- A.unsafeFreezeArr =<< readIORef bufRef+                        k (Just (V.PrimVector oarr 0 osiz))+                        newOutBuffer+                        loop                     else do-                        throwZlibIfMinus_ . withPrimVectorUnsafe dict $ \ pdict off len ->-                            inflate_set_dictionary ps pdict off (fromIntegral len)-                        throwZlibIfMinus (inflate ps (#const Z_FINISH))-                else return r-                oavail :: CUInt <- (#peek struct z_stream_s, avail_out) ps-                return (r', bufSiz - fromIntegral oavail)-            if (r /= (#const Z_STREAM_END) && osiz /= 0)-            then do-                oarr <- A.unsafeFreezeArr buf-                buf' <- A.newPinnedPrimArray bufSiz-                set_avail_out zs buf' bufSiz-                writeIORef bufRef buf'-                zflush finRef zs bufRef (V.PrimVector oarr 0 osiz : acc)-            else do-                oarr <- A.unsafeFreezeArr buf-                let trailing = V.concat . List.reverse $ V.PrimVector oarr 0 osiz : acc-                -- stream ends-                writeIORef finRef True-                if V.null trailing then return Nothing else return (Just trailing)+                        -- stream ends+                        when (osiz /= 0) $ do+                            oarr <- A.unsafeFreezeArr =<< readIORef bufRef+                            k (Just (V.PrimVector oarr 0 osiz))+                        k EOF+            in loop)  -- | Reset decompressor's state so that related 'BIO' can be reused. decompressReset :: ZStream -> IO ()-decompressReset (ZStream fp finRef) = do+decompressReset (ZStream fp) = do     throwZlibIfMinus_ (withCPtr fp inflateReset)-    writeIORef finRef False  -- | Decompress some bytes. decompress :: HasCallStack => DecompressConfig -> V.Bytes -> V.Bytes@@ -342,7 +325,7 @@ throwZlibIfMinus_ = void . throwZlibIfMinus  foreign import ccall unsafe-    create_z_stream :: MBA## (Ptr ZStream) -> IO (Ptr ZStream)+    create_z_stream :: IO (Ptr ZStream)  foreign import ccall unsafe "hs_zlib.c &free_z_stream_inflate"     free_z_stream_inflate :: FunPtr (Ptr ZStream -> IO ())

@@ -14,7 +14,7 @@  module Z.IO.Buffered   ( -- * Input & Output device-    Input(..), Output(..)+    Input(..), Output(..), IODev     -- * Buffered Input   , BufferedInput, bufInput   , newBufferedInput@@ -23,7 +23,7 @@   , unReadBuffer   , clearInputBuffer   , readParser-  , readParseChunks+  , readParseChunk   , readExactly   , readToMagic   , readLine@@ -32,10 +32,13 @@   , BufferedOutput, bufOutput   , newBufferedOutput   , newBufferedOutput'-  , writeBuffer+  , writeBuffer, writeBuffer'   , writeBuilder   , flushBuffer   , clearOutputBuffer+    -- * Buffered Input and Output+  , newBufferedIO+  , newBufferedIO'     -- * common buffer size   , V.defaultChunkSize   , V.smallChunkSize@@ -74,6 +77,15 @@ class Output o where     writeOutput :: o -> Ptr Word8 -> Int -> IO () +-- | Input and Output device+--+-- 'readInput' should return 0 on EOF.+--+-- 'writeOutput' should not return until all data are written (may not+-- necessarily flushed to hardware, that should be done in device specific way).+--+type IODev io = (Input io, Output io)+ -- | Input device with buffer, NOT THREAD SAFE! -- -- * A 'BufferedInput' should not be used in multiple threads, there's no locking mechanism to protect@@ -135,7 +147,19 @@     inputBuffer <- newIORef buf     return (BufferedInput (readInput i) pb inputBuffer) +-- | Open a new buffered input and output with 'V.defaultChunkSize' as buffer size.+newBufferedIO :: IODev dev => dev -> IO (BufferedInput, BufferedOutput)+{-# INLINE newBufferedIO #-}+newBufferedIO dev = newBufferedIO' dev V.defaultChunkSize V.defaultChunkSize +-- | Open a new buffered input and output with given buffer size, e.g. 'V.defaultChunkSize'.+newBufferedIO' :: IODev dev => dev -> Int -> Int -> IO (BufferedInput, BufferedOutput)+{-# INLINE newBufferedIO' #-}+newBufferedIO' dev inSize outSize = do+    i <- newBufferedInput' inSize dev+    o <- newBufferedOutput' outSize dev+    pure (i, o)+ -- | Request bytes chunk from 'BufferedInput'. -- -- The buffering logic is quite simple:@@ -287,19 +311,22 @@ unReadBuffer pb' BufferedInput{..} = unless (V.null pb') $ do     modifyIORef' bufPushBack (\ pb -> pb' `V.append` pb) --- | Read buffer and parse with 'P.ParseChunks'.+-- | Read buffer and parse with 'P.parseChunk' style function. -- -- This function will continuously draw data from input before parsing finish. Unconsumed -- bytes will be returned to buffer. -- -- Throw 'OtherError' with name @EPARSE@ if parsing failed.-readParseChunks :: (T.Print e, HasCallStack) => P.ParseChunks IO V.Bytes e a -> BufferedInput -> IO a-{-# INLINABLE readParseChunks #-}-readParseChunks cp i = do-    bs <- readBuffer i-    (rest, r) <- cp (readBuffer i) bs-    unReadBuffer rest i-    unwrap "EPARSE" r+readParseChunk :: (T.Print e, HasCallStack) => (V.Bytes -> P.Result e a) -> BufferedInput -> IO a+{-# INLINABLE readParseChunk #-}+readParseChunk pc i = loop pc+  where+    loop f = do+        bs <- readBuffer i+        case f bs of+            P.Success v rest -> unReadBuffer rest i >> return v+            P.Failure e rest -> unReadBuffer rest i >> throwOtherError "EPARSE" (T.toText e)+            P.Partial f'     -> loop f'  -- | Read buffer and parse with 'P.Parser'. --@@ -309,7 +336,7 @@ -- Throw 'OtherError' with name @EPARSE@ if parsing failed. readParser :: HasCallStack => P.Parser a -> BufferedInput -> IO a {-# INLINABLE readParser #-}-readParser = readParseChunks . P.parseChunks+readParser = readParseChunk . P.parseChunk  {-| Read until reach a magic bytes, return bytes(including the magic bytes). @@ -397,6 +424,17 @@             copyPrimArray outputBuffer i ba s l   -- copy to buffer             writePrimIORef bufIndex l             -- update index +-- | Write 'V.Bytes' into buffered handle then flush the buffer into output device (if buffer is not empty).+--+-- * If buffer is empty and bytes are larger than half of buffer, directly write bytes,+--   otherwise copy bytes to buffer.+--+-- * If buffer is not empty, then copy bytes to buffer if it can hold, otherwise+--   write buffer first, then try again.+--+writeBuffer' :: HasCallStack => BufferedOutput -> V.Bytes -> IO ()+{-# INLINE writeBuffer' #-}+writeBuffer' bo o = writeBuffer bo o >> flushBuffer bo  -- | Directly write 'B.Builder' into buffered handle. --

@@ -43,7 +43,7 @@   , NoSuchThing(..)   , ResourceBusy(..)   , ResourceExhausted(..)-  , EOF(..)+  , UnexpectedEOF(..)   , IllegalOperation(..)   , PermissionDenied(..)   , UnsatisfiedConstraints(..)@@ -119,7 +119,7 @@ IOE(NoSuchThing) IOE(ResourceBusy) IOE(ResourceExhausted)-IOE(EOF)+IOE(UnexpectedEOF) IOE(IllegalOperation) IOE(PermissionDenied) IOE(UnsatisfiedConstraints)@@ -261,7 +261,7 @@ throwUVError :: CInt -> IOEInfo -> IO a {-# INLINABLE throwUVError #-} throwUVError e info = case e of-    UV_EOF             -> throwIO (EOF                     info)+    UV_EOF             -> throwIO (UnexpectedEOF           info)     UV_E2BIG           -> throwIO (ResourceExhausted       info)     UV_EACCES          -> throwIO (PermissionDenied        info)     UV_EADDRINUSE      -> throwIO (ResourceBusy            info)

@@ -132,7 +132,7 @@ import           Z.IO.Resource import           Z.IO.UV.FFI -#include "_Shared.hs"+#include "fs_shared.hs"  -------------------------------------------------------------------------------- -- File@@ -405,8 +405,8 @@      allocaBytes uvStatSize $ \ s -> do         r <- fromIntegral <$> hs_uv_fs_stat p s         if  | r == UV_ENOENT -> return Nothing-            | r < 0 -> throwUV r-            | otherwise -> Just <$> peekUVStat s+            | r < 0          -> throwUV r+            | otherwise      -> Just <$> peekUVStat s  -- | Equivalent to <http://linux.die.net/man/2/lstat lstat(2)> --@@ -416,8 +416,8 @@      allocaBytes uvStatSize $ \ s -> do         r <- fromIntegral <$> hs_uv_fs_lstat p s         if  | r == UV_ENOENT -> return Nothing-            | r < 0 -> throwUV r-            | otherwise -> Just <$> peekUVStat s+            | r < 0          -> throwUV r+            | otherwise      -> Just <$> peekUVStat s  -- | Equivalent to <http://linux.die.net/man/2/fstat fstat(2)> fstat :: HasCallStack => File -> IO FStat

@@ -145,7 +145,7 @@ import           Z.IO.UV.FFI import           Z.IO.UV.Manager -#include "_Shared.hs"+#include "fs_shared.hs"  -------------------------------------------------------------------------------- -- File@@ -444,8 +444,8 @@             uvm <- getUVManager             withUVRequest' uvm (hs_uv_fs_stat_threaded p s) $ \ r ->                 if  | r == fromIntegral UV_ENOENT -> pure Nothing-                    | r < 0           -> throwUV r-                    | otherwise       -> Just <$> peekUVStat s+                    | r < 0                       -> throwUV r+                    | otherwise                   -> Just <$> peekUVStat s  -- | Equivalent to <http://linux.die.net/man/2/lstat lstat(2)> --@@ -457,8 +457,8 @@             uvm <- getUVManager             withUVRequest' uvm (hs_uv_fs_lstat_threaded p s) $ \ r ->                 if  | r == fromIntegral UV_ENOENT -> pure Nothing-                    | r < 0           -> throwUV r-                    | otherwise       -> Just <$> peekUVStat s+                    | r < 0                       -> throwUV r+                    | otherwise                   -> Just <$> peekUVStat s  -- | Equivalent to <http://linux.die.net/man/2/fstat fstat(2)> fstat :: HasCallStack => File -> IO FStat

@@ -16,7 +16,7 @@     -- dup a file event source     src <- srcf     -- print event to stdout-    runBIO $ src >|> sinkToIO printStd+    runBIO_ $ src . sinkToIO printStd @ -} @@ -31,7 +31,9 @@ import           Data.Bits import qualified Data.HashMap.Strict      as HM import           Data.IORef+#if defined(linux_HOST_OS) import qualified Data.List                as List+#endif import           Data.Primitive.PrimArray import           Data.Word import           GHC.Generics@@ -66,7 +68,7 @@ watchDirs dirs rec callback = do     withResource (initWatchDirs dirs rec) $ \ srcf -> do         src <- srcf-        runBIO $ src >|> sinkToIO callback+        runBIO_ $ src . sinkToIO callback  -- | Start watching a list of given directories, stream version. initWatchDirs :: [CBytes]       -- ^ watching list@@ -108,7 +110,7 @@     cleanUpWatcher mRef sink = do         m <- takeMVar mRef         forM_ m killThread-        void (pull sink)+        void (sink discard EOF)      watchThread mRef dir sink = do         -- IORef store temp events to de-duplicated@@ -205,7 +207,7 @@                 case me of                     Just e -> (Nothing, Just e)                     _      -> (Nothing, Nothing)-            forM_ me' (push sink)+            forM_ me' (stepBIO_ sink)          me' <- atomicModifyIORef' eRef $ \ me ->             case me of@@ -213,4 +215,4 @@                     then (me, Nothing)                     else (Just event, Just e)                 _ -> (Just event, Nothing)-        forM_ me' (push sink)+        forM_ me' (stepBIO_ sink)

@@ -1,156 +0,0 @@--- This file should be included from both base and threaded FS module---- | File bundled with offset.------ Reading or writing using 'Input' \/ 'Output' instance will automatically increase offset.--- 'FilePtr' and its operations are NOT thread safe, use 'MVar' 'FilePtr' in multiple threads.------ The notes on linux 'writeFileP' applied to 'FilePtr' too.-data FilePtr = FilePtr {-# UNPACK #-} !File-                       {-# UNPACK #-} !(PrimIORef Int64)---- |  Create a file offset bundle from an 'File'.----newFilePtr :: File       -- ^ the file we're reading-           -> Int64      -- ^ initial offset-           -> IO FilePtr-newFilePtr uvf off = FilePtr uvf <$> newPrimIORef off---- | Get current offset.-getFilePtrOffset :: FilePtr -> IO Int64-getFilePtrOffset (FilePtr _ offsetRef) = readPrimIORef offsetRef---- | Change current offset.-setFilePtrOffset :: FilePtr -> Int64 -> IO ()-setFilePtrOffset (FilePtr _ offsetRef) = writePrimIORef offsetRef--instance Input FilePtr where-    readInput (FilePtr file offsetRef) buf bufSiz =-        readPrimIORef offsetRef >>= \ off -> do-            l <- readFileP file buf bufSiz off-            writePrimIORef offsetRef (off + fromIntegral l)-            return l--instance Output FilePtr where-    writeOutput (FilePtr file offsetRef) buf bufSiz =-        readPrimIORef offsetRef >>= \ off -> do-            writeFileP file buf bufSiz off-            writePrimIORef offsetRef (off + fromIntegral bufSiz)---- | Quickly open a file and read its content.-readFile :: HasCallStack => CBytes -> IO V.Bytes-readFile filename = do-    withResource (initFile filename O_RDONLY DEFAULT_FILE_MODE) $ \ file -> do-        readAll' =<< newBufferedInput file---- | Quickly open a file and read its content as UTF8 text.-readTextFile :: HasCallStack => CBytes -> IO T.Text-readTextFile filename = T.validate <$> readFile filename---- | Quickly open a file and write some content.-writeFile :: HasCallStack => CBytes -> V.Bytes -> IO ()-writeFile filename content = do-    withResource (initFile filename (O_WRONLY .|. O_CREAT) DEFAULT_FILE_MODE) $ \ file -> do-        withPrimVectorSafe content (writeOutput file)---- | Quickly open a file and write some content as UTF8 text.-writeTextFile :: HasCallStack => CBytes -> T.Text -> IO ()-writeTextFile filename content = writeFile filename (T.getUTF8Bytes content)---- | Quickly open a file and read its content as a JSON value.--- Throw 'OtherError' with name @EPARSE@ if JSON value is not parsed.-readJSONFile :: (HasCallStack, JSON.JSON a) => CBytes -> IO a-readJSONFile filename = unwrap "EPARSE" . JSON.decode' =<< readFile filename---- | Quickly open a file and write a JSON Value.-writeJSONFile :: (HasCallStack, JSON.JSON a) => CBytes -> a -> IO ()-writeJSONFile filename x = writeFile filename (JSON.encode x)-------------------------------------------------------------------------------------- | Find all files and directories within a given directory with a predicator.------ @---  import Z.IO.FileSystem.FilePath (splitExtension)---  -- find all haskell source file within current dir---  scandirRecursively "."  (\\ p _ -> (== ".hs") . snd \<$\> splitExtension p)--- @-scandirRecursively :: HasCallStack => CBytes -> (CBytes -> DirEntType -> IO Bool) -> IO [CBytes]-scandirRecursively dir p = loop [] =<< P.normalize dir-  where-    loop acc0 pdir =-        foldM (\ acc (d,t) -> do-            d' <- pdir `P.join` d-            r <- p d' t-            let acc' = if r then (d':acc) else acc-            if (t == DirEntDir)-            then loop acc' d'-            else return acc'-        ) acc0 =<< scandir pdir-------------------------------------------------------------------------------------- | Does given path exist?----doesPathExist :: CBytes -> IO Bool-doesPathExist path = maybe False (const True) <$> stat' path---- | Returns 'True' if the argument file exists and is either a file or a--- symbolic link to a file, and 'False' otherwise.-doesFileExist :: CBytes -> IO Bool-doesFileExist path = maybe False isFileSt <$> stat' path---- | Returns 'True' if the argument directory exists and is either a directory or a--- symbolic link to a directory, and 'False' otherwise.-doesDirExist :: CBytes -> IO Bool-doesDirExist path = maybe False isDirSt <$> stat' path-------------------------------------------------------------------------------------- | If given path is a symbolic link?-isLink :: HasCallStack => CBytes -> IO Bool-isLink = fmap isLinkSt . lstat---- | If given path is a directory or a symbolic link to a directory?-isDir :: HasCallStack => CBytes -> IO Bool-isDir = fmap isDirSt . stat---- | If given path is a file or a symbolic link to a file?-isFile :: HasCallStack => CBytes -> IO Bool-isFile = fmap isFileSt . stat---- | Shortcut to @\\ st -> stMode st .&. S_IFMT == S_IFLNK@------ Note you should use 'lstat' to get the link's stat.-isLinkSt :: FStat -> Bool-isLinkSt st = stMode st .&. S_IFMT == S_IFLNK---- | Shortcut to @\\ st -> stMode st .&. S_IFMT == S_IFDIR@-isDirSt :: FStat -> Bool-isDirSt st = stMode st .&. S_IFMT == S_IFDIR---- | Shortcut to @\\ st -> stMode st .&. S_IFMT == S_IFREG@-isFileSt :: FStat -> Bool-isFileSt st = stMode st .&. S_IFMT == S_IFREG------------------------------------------------------------------------------------- | Make a temporary file under system 'Env.getTempDir' and automatically clean after used.------ >>> withResource (initTempFile "foo") $ printStd--- File 13----initTempFile :: CBytes -> Resource File-initTempFile prefix =-    initResource initAction unlink >>= (\f -> initFile f O_RDWR DEFAULT_FILE_MODE)-    where-        initAction = Env.getTempDir >>= (`P.join` prefix) >>= mkstemp---- | Make a temporary directory under system 'Env.getTempDir' and automatically clean after used.------ >>> withResource (initTempDir "foo") $ printStd--- "/tmp/fooxfWR0L"----initTempDir :: CBytes -> Resource CBytes-initTempDir prefix =-    initResource (Env.getTempDir >>= (`P.join` prefix) >>= mkdtemp) rmrf

@@ -28,7 +28,7 @@ import           Foreign.C.Types import           Foreign.Marshal.Utils import           Foreign.Ptr-import           Foreign.Storable +import           Foreign.Storable import           GHC.Generics import           Z.Data.CBytes              as CBytes import           Z.Data.Text.Print          (Print(..))@@ -321,7 +321,7 @@ getNameInfo flags doHost doService addr = withUVInitDo $ do     (host, (service, _)) <- allocCBytes (fromIntegral h_len) $ \ ptr_h ->         allocCBytes (fromIntegral s_len) $ \ ptr_s ->-        withSocketAddr addr $ \ ptr_addr -> +        withSocketAddr addr $ \ ptr_addr ->             throwUVIfMinus_ $ hs_getnameinfo ptr_addr addr_len ptr_h h_len ptr_s s_len cflag     return (host, service)   where@@ -357,7 +357,7 @@  ----------------------------------------------------------------------------- foreign import ccall safe "hs_getaddrinfo"-    hs_getaddrinfo :: Ptr Word8 -- ^ host +    hs_getaddrinfo :: Ptr Word8 -- ^ host                    -> Ptr Word8 -- ^ service                    -> Ptr AddrInfo   -- ^ hints                    -> Ptr (Ptr AddrInfo) -- ^ output addrinfo linked list@@ -368,7 +368,7 @@ foreign import ccall safe "hs_getnameinfo"     hs_getnameinfo :: Ptr SocketAddr                       -> CSize-                      -> CString -- ^ output host +                      -> CString -- ^ output host                       -> CSize                       -> CString -- ^ output service                       -> CSize

@@ -44,6 +44,12 @@   -- * port numbber   , PortNumber(..)   , portAny+  , defaultPortNumberHTTP+  , defaultPortNumberHTTPS+  , defaultPortNumberSMTP+  , defaultPortNumberPOP3+  , defaultPortNumberIMAP+  , defaultPortNumberIRC   -- * family, type, protocol   -- ** SocketFamily   , SocketFamily@@ -90,7 +96,7 @@ import           Z.IO.Exception import           Z.Foreign -#include "hs_uv.h" +#include "hs_uv.h"  #if defined(i386_HOST_ARCH) && defined(mingw32_HOST_OS) #let CALLCONV = "stdcall"@@ -109,7 +115,7 @@ #endif  -- | IPv4 or IPv6 socket address, i.e. the `sockaddr_in` or `sockaddr_in6` struct.--- +-- -- Example on JSON instance: -- -- @@@ -118,7 +124,7 @@ -- > JSON.encodeText  $ ipv4 "128.14.32.1" 9090 -- "{\"addr\":[128,14,32,1],\"port\":9090}" -- @-data SocketAddr +data SocketAddr     = SocketAddrIPv4         {-# UNPACK #-} !IPv4    -- sin_addr  (ditto)         {-# UNPACK #-} !PortNumber  -- sin_port  (network byte order)@@ -129,7 +135,7 @@         {-# UNPACK #-} !ScopeID     -- sin6_scope_id (ditto)     deriving (Eq, Ord, Generic) -instance JSON SocketAddr where +instance JSON SocketAddr where     {-# INLINE encodeJSON #-}     encodeJSON (SocketAddrIPv4 addr port) = T.curly $ do         "addr" `B.kv` encodeJSON addr@@ -145,11 +151,11 @@         "scope" `B.kv` encodeJSON scope      {-# INLINE toValue #-}-    toValue (SocketAddrIPv4 addr port) = JSON.Object . V.pack $ +    toValue (SocketAddrIPv4 addr port) = JSON.Object . V.pack $         [ ("addr", toValue addr)         , ("number", toValue port)         ]-    toValue (SocketAddrIPv6 addr port flow scope) = JSON.Object . V.pack $ +    toValue (SocketAddrIPv6 addr port flow scope) = JSON.Object . V.pack $         [ ("addr", toValue addr)         , ("number", toValue port)         , ("flow", toValue flow)@@ -189,7 +195,7 @@        = T.toUTF8Builder addr >> T.char7 ':' >> T.toUTF8Builder port     toUTF8BuilderP _ (SocketAddrIPv6 addr port _ _) = do         T.square (T.toUTF8Builder addr)-        T.char7 ':' +        T.char7 ':'         T.toUTF8Builder port  sockAddrFamily :: SocketAddr -> SocketFamily@@ -222,7 +228,7 @@ -- JSON instance encode ipv4 address into an array with 4 'Word8' octets. newtype IPv4 = IPv4 { getIPv4Addr :: Word32 }     deriving (Eq, Ord, Generic)-    + instance JSON IPv4 where     {-# INLINE encodeJSON #-}     encodeJSON = encodeJSON . ipv4AddrToTuple@@ -235,13 +241,13 @@ instance Print IPv4 where     toUTF8BuilderP _ ia = do         let (a,b,c,d) = ipv4AddrToTuple ia-        T.int a -        T.char7 '.' -        T.int b  -        T.char7 '.' +        T.int a+        T.char7 '.'+        T.int b+        T.char7 '.'         T.int c-        T.char7 '.' -        T.int d  +        T.char7 '.'+        T.int d  -- | @0.0.0.0@ ipv4Any             :: IPv4@@ -273,7 +279,7 @@  instance Storable IPv4 where     sizeOf _ = 4-    alignment _ = alignment (undefined :: Word32) +    alignment _ = alignment (undefined :: Word32)     peek p = (IPv4 . ntohl) `fmap` peekByteOff p 0     poke p (IPv4 ia) = pokeByteOff p 0 (htonl ia) @@ -282,7 +288,7 @@     pokeMBA p off x = pokeMBA p off (htonl (getIPv4Addr x))     peekMBA p off = IPv4 . ntohl <$> peekMBA p off     indexBA p off = IPv4 (ntohl (indexBA p off))-    + -- | Converts 'IPv4' to representation-independent IPv4 quadruple. -- For example for @127.0.0.1@ the function will return @(127, 0, 0, 1)@ -- regardless of host endianness.@@ -307,7 +313,7 @@ data IPv6 = IPv6 {-# UNPACK #-}!Word32                  {-# UNPACK #-}!Word32                  {-# UNPACK #-}!Word32-                 {-# UNPACK #-}!Word32 +                 {-# UNPACK #-}!Word32     deriving (Eq, Ord, Generic)  instance JSON IPv6 where@@ -415,7 +421,7 @@ instance Unaligned IPv6 where     unalignedSize = (#size struct in6_addr) -    indexBA p off = +    indexBA p off =         let a = indexBA p (off + s6_addr_offset + 0)             b = indexBA p  (off + s6_addr_offset + 4)             c = indexBA p  (off + s6_addr_offset + 8)@@ -489,8 +495,8 @@         (#size struct sockaddr_in)         (#alignment struct sockaddr_in) $ \ p -> pokeSocketAddr p sa >> f p withSocketAddr sa@(SocketAddrIPv6 _ _ _ _) f = do-    allocaBytesAligned -        (#size struct sockaddr_in6) +    allocaBytesAligned+        (#size struct sockaddr_in6)         (#alignment struct sockaddr_in6) $ \ p -> pokeSocketAddr p sa >> f p  -- | Pass 'SocketAddr' to FFI as pointer.@@ -499,12 +505,12 @@ -- withSocketAddrUnsafe :: SocketAddr -> (MBA## SocketAddr -> IO a) -> IO a withSocketAddrUnsafe sa@(SocketAddrIPv4 _ _) f = do-    (MutableByteArray p) <- newByteArray (#size struct sockaddr_in) -    pokeSocketAddrMBA p sa +    (MutableByteArray p) <- newByteArray (#size struct sockaddr_in)+    pokeSocketAddrMBA p sa     f p withSocketAddrUnsafe sa@(SocketAddrIPv6 _ _ _ _) f = do-    (MutableByteArray p) <- newByteArray (#size struct sockaddr_in6) -    pokeSocketAddrMBA p sa +    (MutableByteArray p) <- newByteArray (#size struct sockaddr_in6)+    pokeSocketAddrMBA p sa     f p  sizeOfSocketAddr :: SocketAddr -> CSize@@ -524,7 +530,7 @@ -- withSocketAddrStorageUnsafe :: (MBA## SocketAddr -> IO ()) -> IO SocketAddr withSocketAddrStorageUnsafe f = do-    (MutableByteArray p) <- newByteArray (#size struct sockaddr_storage) +    (MutableByteArray p) <- newByteArray (#size struct sockaddr_storage)     f p     peekSocketAddrMBA p @@ -536,14 +542,14 @@     family <- peekMBA p (#offset struct sockaddr, sa_family)     case family :: CSaFamily of         (#const AF_INET) -> do-            addr <- peekMBA p (#offset struct sockaddr_in, sin_addr) -            port <- peekMBA p (#offset struct sockaddr_in, sin_port) +            addr <- peekMBA p (#offset struct sockaddr_in, sin_addr)+            port <- peekMBA p (#offset struct sockaddr_in, sin_port)             return (SocketAddrIPv4 addr port)         (#const AF_INET6) -> do-            port <- peekMBA p (#offset struct sockaddr_in6, sin6_port) -            flow <- peekMBA p (#offset struct sockaddr_in6, sin6_flowinfo) -            addr <- peekMBA p (#offset struct sockaddr_in6, sin6_addr) -            scope <- peekMBA p (#offset struct sockaddr_in6, sin6_scope_id) +            port <- peekMBA p (#offset struct sockaddr_in6, sin6_port)+            flow <- peekMBA p (#offset struct sockaddr_in6, sin6_flowinfo)+            addr <- peekMBA p (#offset struct sockaddr_in6, sin6_addr)+            scope <- peekMBA p (#offset struct sockaddr_in6, sin6_scope_id)             return (SocketAddrIPv6 addr port flow scope)         _ -> do let errno = UV_EAI_ADDRFAMILY                 name <- uvErrName errno@@ -578,7 +584,7 @@ -- Port Numbers  -- | Port number.--- +-- -- Use the @Num@ instance (i.e. use a literal) to create a --   @PortNumber@ value. -- -- >>> 1 :: PortNumber@@ -593,7 +599,7 @@ -- True -- >>> 50000 + (10000 :: PortNumber) -- 60000-newtype PortNumber = PortNumber Word16 +newtype PortNumber = PortNumber Word16     deriving (Eq, Ord, Enum, Generic)     deriving newtype (Show, Print, Read, Num, Bounded, Real, Integral, JSON) @@ -601,6 +607,24 @@ portAny :: PortNumber portAny = PortNumber 0 +defaultPortNumberHTTP :: PortNumber+defaultPortNumberHTTP = 80++defaultPortNumberHTTPS :: PortNumber+defaultPortNumberHTTPS = 443++defaultPortNumberSMTP :: PortNumber+defaultPortNumberSMTP = 25++defaultPortNumberPOP3 :: PortNumber+defaultPortNumberPOP3 = 110++defaultPortNumberIMAP :: PortNumber+defaultPortNumberIMAP = 143++defaultPortNumberIRC :: PortNumber+defaultPortNumberIRC = 194+ instance Storable PortNumber where    sizeOf    _ = sizeOf    (0 :: Word16)    alignment _ = alignment (0 :: Word16)@@ -612,7 +636,7 @@    indexBA p off = PortNumber . ntohs $ indexBA p off    pokeMBA p off (PortNumber po) = pokeMBA p off (htons po)    peekMBA p off = PortNumber . ntohs <$> peekMBA p off-    + --------------------------------------------------------------------------------  type SocketFamily = CInt

@@ -129,8 +129,8 @@ -- | Create a new resource and run some computation, resource is guarantee to -- be closed. ----- Be care don't leak the resource through computation return value, because--- after the computation finishes, the resource is closed already.+-- Be careful, don't leak the resource through the computation return value+-- because after the computation finishes, the resource is already closed. -- withResource :: (MonadCatch.MonadMask m, MonadIO m, HasCallStack)              => Resource a -> (a -> m b) -> m b

@@ -32,9 +32,8 @@ #include <stdlib.h> #include <HsFFI.h> -z_stream* create_z_stream(z_stream** streamp){+z_stream* create_z_stream(){     z_stream* stream = malloc(sizeof(z_stream));-    *streamp = stream;     if (stream) {         stream->zalloc = Z_NULL;         stream->zfree = Z_NULL;

@@ -0,0 +1,156 @@+-- This file should be included from both base and threaded FS module++-- | File bundled with offset.+--+-- Reading or writing using 'Input' \/ 'Output' instance will automatically increase offset.+-- 'FilePtr' and its operations are NOT thread safe, use 'MVar' 'FilePtr' in multiple threads.+--+-- The notes on linux 'writeFileP' applied to 'FilePtr' too.+data FilePtr = FilePtr {-# UNPACK #-} !File+                       {-# UNPACK #-} !(PrimIORef Int64)++-- |  Create a file offset bundle from an 'File'.+--+newFilePtr :: File       -- ^ the file we're reading+           -> Int64      -- ^ initial offset+           -> IO FilePtr+newFilePtr uvf off = FilePtr uvf <$> newPrimIORef off++-- | Get current offset.+getFilePtrOffset :: FilePtr -> IO Int64+getFilePtrOffset (FilePtr _ offsetRef) = readPrimIORef offsetRef++-- | Change current offset.+setFilePtrOffset :: FilePtr -> Int64 -> IO ()+setFilePtrOffset (FilePtr _ offsetRef) = writePrimIORef offsetRef++instance Input FilePtr where+    readInput (FilePtr file offsetRef) buf bufSiz =+        readPrimIORef offsetRef >>= \ off -> do+            l <- readFileP file buf bufSiz off+            writePrimIORef offsetRef (off + fromIntegral l)+            return l++instance Output FilePtr where+    writeOutput (FilePtr file offsetRef) buf bufSiz =+        readPrimIORef offsetRef >>= \ off -> do+            writeFileP file buf bufSiz off+            writePrimIORef offsetRef (off + fromIntegral bufSiz)++-- | Quickly open a file and read its content.+readFile :: HasCallStack => CBytes -> IO V.Bytes+readFile filename = do+    withResource (initFile filename O_RDONLY DEFAULT_FILE_MODE) $ \ file -> do+        readAll' =<< newBufferedInput file++-- | Quickly open a file and read its content as UTF8 text.+readTextFile :: HasCallStack => CBytes -> IO T.Text+readTextFile filename = T.validate <$> readFile filename++-- | Quickly open a file and write some content.+writeFile :: HasCallStack => CBytes -> V.Bytes -> IO ()+writeFile filename content = do+    withResource (initFile filename (O_WRONLY .|. O_CREAT) DEFAULT_FILE_MODE) $ \ file -> do+        withPrimVectorSafe content (writeOutput file)++-- | Quickly open a file and write some content as UTF8 text.+writeTextFile :: HasCallStack => CBytes -> T.Text -> IO ()+writeTextFile filename content = writeFile filename (T.getUTF8Bytes content)++-- | Quickly open a file and read its content as a JSON value.+-- Throw 'OtherError' with name @EPARSE@ if JSON value is not parsed.+readJSONFile :: (HasCallStack, JSON.JSON a) => CBytes -> IO a+readJSONFile filename = unwrap "EPARSE" . JSON.decode' =<< readFile filename++-- | Quickly open a file and write a JSON Value.+writeJSONFile :: (HasCallStack, JSON.JSON a) => CBytes -> a -> IO ()+writeJSONFile filename x = writeFile filename (JSON.encode x)++--------------------------------------------------------------------------------++-- | Find all files and directories within a given directory with a predicator.+--+-- @+--  import Z.IO.FileSystem.FilePath (splitExtension)+--  -- find all haskell source file within current dir+--  scandirRecursively "."  (\\ p _ -> (== ".hs") . snd \<$\> splitExtension p)+-- @+scandirRecursively :: HasCallStack => CBytes -> (CBytes -> DirEntType -> IO Bool) -> IO [CBytes]+scandirRecursively dir p = loop [] =<< P.normalize dir+  where+    loop acc0 pdir =+        foldM (\ acc (d,t) -> do+            d' <- pdir `P.join` d+            r <- p d' t+            let acc' = if r then (d':acc) else acc+            if (t == DirEntDir)+            then loop acc' d'+            else return acc'+        ) acc0 =<< scandir pdir++--------------------------------------------------------------------------------++-- | Does given path exist?+--+doesPathExist :: CBytes -> IO Bool+doesPathExist path = maybe False (const True) <$> stat' path++-- | Returns 'True' if the argument file exists and is either a file or a+-- symbolic link to a file, and 'False' otherwise.+doesFileExist :: CBytes -> IO Bool+doesFileExist path = maybe False isFileSt <$> stat' path++-- | Returns 'True' if the argument directory exists and is either a directory or a+-- symbolic link to a directory, and 'False' otherwise.+doesDirExist :: CBytes -> IO Bool+doesDirExist path = maybe False isDirSt <$> stat' path++--------------------------------------------------------------------------------++-- | If given path is a symbolic link?+isLink :: HasCallStack => CBytes -> IO Bool+isLink = fmap isLinkSt . lstat++-- | If given path is a directory or a symbolic link to a directory?+isDir :: HasCallStack => CBytes -> IO Bool+isDir = fmap isDirSt . stat++-- | If given path is a file or a symbolic link to a file?+isFile :: HasCallStack => CBytes -> IO Bool+isFile = fmap isFileSt . stat++-- | Shortcut to @\\ st -> stMode st .&. S_IFMT == S_IFLNK@+--+-- Note you should use 'lstat' to get the link's stat.+isLinkSt :: FStat -> Bool+isLinkSt st = stMode st .&. S_IFMT == S_IFLNK++-- | Shortcut to @\\ st -> stMode st .&. S_IFMT == S_IFDIR@+isDirSt :: FStat -> Bool+isDirSt st = stMode st .&. S_IFMT == S_IFDIR++-- | Shortcut to @\\ st -> stMode st .&. S_IFMT == S_IFREG@+isFileSt :: FStat -> Bool+isFileSt st = stMode st .&. S_IFMT == S_IFREG++-------------------------------------------------------------------------------++-- | Make a temporary file under system 'Env.getTempDir' and automatically clean after used.+--+-- >>> withResource (initTempFile "foo") $ printStd+-- File 13+--+initTempFile :: CBytes -> Resource File+initTempFile prefix =+    initResource initAction unlink >>= (\f -> initFile f O_RDWR DEFAULT_FILE_MODE)+    where+        initAction = Env.getTempDir >>= (`P.join` prefix) >>= mkstemp++-- | Make a temporary directory under system 'Env.getTempDir' and automatically clean after used.+--+-- >>> withResource (initTempDir "foo") $ printStd+-- "/tmp/fooxfWR0L"+--+initTempDir :: CBytes -> Resource CBytes+initTempDir prefix =+    initResource (Env.getTempDir >>= (`P.join` prefix) >>= mkdtemp) rmrf

@@ -1,15 +1,18 @@ {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}  module Z.IO.BIO.ConcurrentSpec where  import           Control.Concurrent import           Control.Monad+import           Control.Monad.IO.Class import           Data.IORef import           Z.IO.BIO.Concurrent import           Z.IO.BIO import           Test.QuickCheck import           Test.QuickCheck.Function import           Test.QuickCheck.Property+import           Test.QuickCheck.Monadic as QM import           Test.Hspec import           Test.Hspec.QuickCheck import           Test.HUnit@@ -19,6 +22,10 @@ spec :: Spec spec = describe "BIO.Concurrent" $ do +    prop "zipBIO works like zip on Sources" $ \ xs ys -> monadicIO $ do+        (zs :: [(Int, Int)]) <- liftIO . runBIO $ zipBIO  (sourceFromList xs) (sourceFromList ys)+        QM.assert (zs == zip xs ys)+     it "TQueueNode works as expected" $ do         let content = [1..1000] @@ -27,12 +34,12 @@          let producter = do                 src' <- sourceListWithDelay content-                runBIO (src' >|> sink)+                runBIO_ (src' . sink)          let consumer =  do                 (rRef, sink') <- sinkToList-                runBIO (src >|> sink')-                r <- readIORef rRef+                runBIO_ (src . sink')+                r <- takeMVar rRef                 atomicModifyIORef' sumRef $ \ x -> (x + sum r, ())          forkIO $ consumer@@ -54,12 +61,12 @@          let producter = do                 src' <- sourceListWithDelay content-                runBIO (src' >|> sink)+                runBIO_ (src' . sink)          let consumer =  do                 (rRef, sink') <- sinkToList-                runBIO (src >|> sink')-                r <- readIORef rRef+                runBIO_ (src . sink')+                r <- takeMVar rRef                 atomicModifyIORef' sumRef $ \ x -> (x + sum r, ())          forkIO $ consumer@@ -81,13 +88,13 @@          let producter = do                 src' <- sourceListWithDelay content-                runBIO (src' >|> sink)+                runBIO_ (src' . sink)          let consumer =  do                 (rRef, sink') <- sinkToList                 src <- srcf-                runBIO (src >|> sink')-                r <- readIORef rRef+                runBIO_ (src . sink')+                r <- takeMVar rRef                 atomicModifyIORef' sumRef $ \ x -> (x + sum r, ())          forkIO $ consumer@@ -103,15 +110,9 @@   sourceListWithDelay :: [Int] -> IO (Source Int)-sourceListWithDelay xs0 = do-    xsRef <- newIORef xs0-    return BIO{ pull = popper xsRef }-  where-    popper xsRef = do-        xs <- readIORef xsRef-        case xs of-            (x:xs') -> do-                writeIORef xsRef xs'-                threadDelay x-                return (Just x)-            _ -> return Nothing+sourceListWithDelay xs = do+    return $ \ k _ -> do+        forM_ xs $ \ x -> do+            threadDelay x+            k (Just x)+        k Nothing

@@ -48,22 +48,22 @@              B.concat vs @=? BL.toStrict vs' -        prop "compress >|> decompress" $ \ xss -> do+        prop "compress . decompress" $ \ xss -> do             (_, c) <- newCompress defaultCompressConfig             (_, d) <- newDecompress defaultDecompressConfig              let vs = Prelude.map V.pack xss-            vs' <- runBlocks (c >|> d) vs+            vs' <- runBlocks (c . d) vs              V.concat vs @=? V.concat vs' -        prop "compress >|> decompress (with dict)" $ \ xss -> do+        prop "compress . decompress (with dict)" $ \ xss -> do             let dict = "aabbccdd"              (_, c) <- newCompress defaultCompressConfig{compressDictionary = dict}             (_, d) <- newDecompress defaultDecompressConfig{decompressDictionary = dict}              let vs = Prelude.map V.pack xss-            vs' <- runBlocks (c >|> d) vs+            vs' <- runBlocks (c . d) vs              V.concat vs @=? V.concat vs'

@@ -2,6 +2,7 @@  module Z.IO.BIOSpec where +import           Control.Concurrent import           Control.Monad import qualified Codec.Compression.Zlib as TheZlib import           Data.IORef@@ -23,22 +24,22 @@     describe "decode . encode === id(Base64)" $         prop "Base64" $ \ xs ->             let r = unsafePerformIO $ do-                    src <- sourceFromList xs+                    let src = sourceFromList xs                     (rRef, sink) <- sinkToList                     enc <- newBase64Encoder                     dec <- newBase64Decoder-                    runBIO $ src >|> enc >|> dec >|> sink-                    readIORef rRef+                    runBIO $ src . enc . dec . sink+                    takeMVar rRef             in V.concat r === V.concat xs      describe "decode . encode === id(Hex)" $ do         prop "Hex" $ \ xs upper ->             let r = unsafePerformIO $ do-                    src <- sourceFromList xs+                    let src = sourceFromList xs                     (rRef, sink) <- sinkToList                     let enc = hexEncoder upper                     dec <- newHexDecoder-                    runBIO $ src >|> enc >|> dec >|> sink-                    readIORef rRef+                    runBIO $ src . enc . dec . sink+                    takeMVar rRef             in V.concat r === V.concat xs