binary-io 0.2.0 → 0.3.0

raw patch · 7 files changed

+146/−28 lines, 7 filesdep +asyncdep +dequedep +stmdep ~hspecPVP ok

version bump matches the API change (PVP)

Dependencies added: async, deque, stm

Dependency ranges changed: hspec

API changes (from Hackage documentation)

Files

• ChangeLog.md +5/−0
• binary-io.cabal +10/−4
• lib/Data/Binary/IO.hs +40/−22
• lib/Data/Binary/IO/Internal/AwaitNotify.hs +41/−0
• test/Data/Binary/IO/Internal/AwaitNotifySpec.hs +37/−0
• test/Data/Binary/IOSpec.hs +7/−1
• test/Main.hs +6/−1

@@ -1,5 +1,10 @@ # binary-io +## 0.3.0++* Add isEmpty function to check if a read source has no more input available+* Enhance newPipe to produce a Reader that detects when its paired Writer went out-of-scope+ ## 0.2.0  * Add newPipe function that can create a connected Reader and Writer

@@ -5,7 +5,7 @@   binary-io  version:-  0.2.0+  0.3.0  category:   Data, Parsing, IO@@ -53,13 +53,16 @@   build-depends:     base == 4.*,     bytestring,-    binary >= 0.7.2+    binary >= 0.7.2,+    process,+    deque    hs-source-dirs:     lib    exposed-modules:     Data.Binary.IO+    Data.Binary.IO.Internal.AwaitNotify  test-suite binary-io-tests   type:@@ -69,7 +72,7 @@     Haskell2010    ghc-options:-    -Wall -Wextra -Wno-name-shadowing+    -Wall -Wextra -Wno-name-shadowing -threaded -with-rtsopts=-N    build-depends:     base,@@ -77,13 +80,16 @@     binary-io,     bytestring,     process,-    hspec+    hspec >= 2.7.1,+    async,+    stm    hs-source-dirs:     test    other-modules:     Data.Binary.IOSpec+    Data.Binary.IO.Internal.AwaitNotifySpec    main-is:     Main.hs

@@ -25,6 +25,7 @@     -- * Classes   , CanGet (..)   , read+  , isEmpty    , CanPut (..)   , write@@ -33,21 +34,22 @@  import Prelude hiding (read) -import qualified Control.Concurrent.Chan as Chan import           Control.Concurrent.MVar (MVar, modifyMVar, newMVar)-import qualified Control.Concurrent.MVar as MVar import qualified Control.Exception as Exception-import           Control.Monad (unless, void)+import           Control.Monad (join, unless) import           Data.Bifunctor (bimap) import qualified Data.Binary as Binary import qualified Data.Binary.Get as Binary.Get+import           Data.Binary.IO.Internal.AwaitNotify (newAwaitNotify, runAwait, runNotify) import qualified Data.Binary.Put as Binary.Put import qualified Data.ByteString as ByteString.Strict import qualified Data.ByteString.Lazy as ByteString import           Data.ByteString.Lazy.Internal (ByteString (Chunk, Empty))+import           Data.IORef (atomicModifyIORef', mkWeakIORef, newIORef)+import qualified Deque.Strict as Deque import           System.IO (Handle, hSetBinaryMode) import           System.IO.Unsafe (unsafeInterleaveIO)-import qualified System.Mem.Weak as Weak+import           System.Mem.Weak (deRefWeak)  -- * Reader @@ -137,6 +139,8 @@   Reader <$> newMVar posReader  -- | This function works very similar to 'newReader' except no 'Handle' is involved.+-- The chunk producers indicates the end of the stream by returning an empty+-- 'ByteString.Strict.ByteString'. -- -- @since 0.1.1 newReaderWith@@ -183,19 +187,26 @@ -- @since 0.2.0 newPipe :: IO (Reader, Writer) newPipe = do-  chan <- Chan.newChan-  mvar <- MVar.newMVar chan--  Weak.addFinalizer chan (void (MVar.tryTakeMVar mvar))+  chan <- newIORef mempty+  weakChan <- mkWeakIORef chan (pure ())+  (await, notify) <- newAwaitNotify    let     read = do-      mbChan <- MVar.tryReadMVar mvar-      maybe (pure ByteString.Strict.empty) Chan.readChan mbChan+      mbChan <- deRefWeak weakChan+      case mbChan of+        Nothing -> pure ByteString.Strict.empty+        Just chan -> join $+          atomicModifyIORef' chan $ \queue ->+            case Deque.uncons queue of+              Just (elem, queue) -> (queue, pure elem)+              Nothing -> (queue, runAwait await >> read)      write msg =-      unless (ByteString.Strict.null msg) $-        Chan.writeChan chan msg+      unless (ByteString.Strict.null msg) $ do+        atomicModifyIORef' chan $ \queue ->+          (Deque.snoc msg queue, ())+        runNotify notify    reader <- newReaderWith read   let writer = newWriterWith write@@ -249,6 +260,23 @@ instance CanGet Duplex where   runGet = runGet . duplexReader +-- | Read something from @r@.+--+-- @since 0.0.1+read+  :: (CanGet r, Binary.Binary a)+  => r -- ^ Read source+  -> IO a+read reader =+  runGet reader Binary.get++-- | Check if there is no more input to consume. This function may block. All properties of 'runGet'+-- apply to this function as well.+--+-- @since 0.3.0+isEmpty :: CanGet r => r -> IO Bool+isEmpty reader = runGet reader Binary.Get.isEmpty+ -- | @w@ can execute 'Binary.Put' operations -- -- @since 0.0.1@@ -267,16 +295,6 @@  instance CanPut Duplex where   runPut = runPut . duplexWriter---- | Read something from @r@.------ @since 0.0.1-read-  :: (CanGet r, Binary.Binary a)-  => r -- ^ Read source-  -> IO a-read reader =-  runGet reader Binary.get  -- | Write something to @w@. --

@@ -0,0 +1,41 @@+{-# LANGUAGE TypeApplications #-}++{-# OPTIONS_HADDOCK hide #-}++module Data.Binary.IO.Internal.AwaitNotify+  ( Await (..)+  , Notify (..)+  , newAwaitNotify+  )+where++import qualified Foreign++import           Data.Word (Word8)+import qualified System.IO as IO+import qualified System.Process as Process++-- | Await signal from a paired 'Notify'. Returns 'False' if the paired 'Notify' does not exist+-- (any more).+newtype Await = Await+  { runAwait :: IO Bool }++-- | Notify the paired 'Await'.+newtype Notify = Notify+  { runNotify :: IO () }++newAwaitNotify :: IO (Await, Notify)+newAwaitNotify = do+  buf <- Foreign.calloc @Word8+  (read, write) <- Process.createPipe++  IO.hSetBuffering read IO.NoBuffering+  IO.hSetBuffering write IO.NoBuffering++  IO.hSetBinaryMode read True+  IO.hSetBinaryMode write True++  let notify = IO.hPutBuf write buf 1+  let await  = (> 0) <$> IO.hGetBufSome read buf 1++  pure (Await await, Notify notify)

@@ -0,0 +1,37 @@+{-# LANGUAGE NumericUnderscores #-}++module Data.Binary.IO.Internal.AwaitNotifySpec (spec) where++import           Control.Concurrent (forkIO, threadDelay)+import           Data.Binary.IO.Internal.AwaitNotify+import qualified Test.Hspec as Hspec++spec :: Hspec.Spec+spec =+  Hspec.describe "newAwaitNotify" $ do+    Hspec.it "can get notified single-threadedly" $ do+      (await, notify) <- newAwaitNotify++      runNotify notify++      result <- runAwait await+      result `Hspec.shouldBe` True++    Hspec.it "can get notified multi-threadedly" $ do+      (await, notify) <- newAwaitNotify++      _ <- forkIO $ do+        threadDelay 1_000_000+        runNotify notify++      result <- runAwait await+      result `Hspec.shouldBe` True++    Hspec.it "detects that Notify went out-of-scope" $ do+      (await, _notify) <- newAwaitNotify+      result <- runAwait await+      result `Hspec.shouldBe` False++    Hspec.it "can notify without a paired " $ do+      (_await, notify) <- newAwaitNotify+      runNotify notify

@@ -184,7 +184,7 @@          Hspec.shouldThrow (write writer "Hello World") isIllegalOperation -  Hspec.describe "Pipe" $+  Hspec.describe "Pipe" $ do     Hspec.it "is connected" $ do       (reader, writer) <- newPipe @@ -192,3 +192,9 @@       "Hello World" <- read reader        pure ()++    Hspec.it "ends if writer is out of scope" $ do+      (reader, _writer) <- newPipe++      isEmpty <- isEmpty reader+      Hspec.shouldBe isEmpty True

@@ -2,7 +2,12 @@  import Test.Hspec +import qualified Data.Binary.IO.Internal.AwaitNotifySpec import qualified Data.Binary.IOSpec  main :: IO ()-main = hspec Data.Binary.IOSpec.spec+main = hspec $ do+  describe "Data" $ describe "Binary" $ describe "IO" $ do+    describe "Internal" $ describe "AwaitNotify"+      Data.Binary.IO.Internal.AwaitNotifySpec.spec+    Data.Binary.IOSpec.spec