binary-io 0.3.0 → 0.4.0

raw patch · 7 files changed

+474/−219 lines, 7 filesdep +concurrencydep +exceptionsdep +transformersPVP ok

version bump matches the API change (PVP)

Dependencies added: concurrency, exceptions, transformers

API changes (from Hackage documentation)

Files

• ChangeLog.md +7/−0
• binary-io.cabal +8/−4
• lib/Data/Binary/IO.hs +87/−202
• lib/Data/Binary/IO/Internal/AwaitNotify.hs +11/−5
• lib/Data/Binary/IO/Lifted.hs +354/−0
• test/Data/Binary/IO/Internal/AwaitNotifySpec.hs +1/−1
• test/Data/Binary/IOSpec.hs +6/−7

@@ -1,5 +1,12 @@ # binary-io +## 0.4.0++* Add module Data.Binary.IO.Lifted module which contains versions of all functions and types+  for the use in functors other than IO+* Make all functions and types in Data.Binary.IO aliases of the ones in the Lifted module+* Reduce allocations for Await-Notify pairs+ ## 0.3.0  * Add isEmpty function to check if a read source has no more input available

@@ -5,7 +5,7 @@   binary-io  version:-  0.3.0+  0.4.0  category:   Data, Parsing, IO@@ -16,7 +16,7 @@ description:   Read and write values of types that implement Binary   .-  See module "Data.Binary.IO".+  See module "Data.Binary.IO.Lifted".  author:   Ole Krüger <haskell-binary-io@vprsm.de>@@ -48,20 +48,24 @@     Haskell2010    ghc-options:-    -Wall -Wextra -Wno-name-shadowing+    -Wall -Wextra -Wno-name-shadowing -Wredundant-constraints    build-depends:     base == 4.*,     bytestring,     binary >= 0.7.2,     process,-    deque+    deque,+    transformers,+    concurrency,+    exceptions    hs-source-dirs:     lib    exposed-modules:     Data.Binary.IO+    Data.Binary.IO.Lifted     Data.Binary.IO.Internal.AwaitNotify  test-suite binary-io-tests

@@ -1,16 +1,26 @@-{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE DerivingVia #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MonoLocalBinds #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE UndecidableInstances #-}  -- | Read and write values of types that implement 'Binary.Binary' from and to 'Handle's+--+-- This module homes the unlifted API variant. For proper documentation check out the equally named+-- functions in "Data.Binary.IO.Lifted"+-- module Data.Binary.IO-  ( -- * Readers-    ReaderError (..)+  ( -- * Reader+    Lifted.ReaderError (..) -  , Reader+  , Reader (..)   , newReader   , newReaderWith -    -- * Writers-  , Writer+    -- * Writer+  , Writer (..)   , newWriter   , newWriterWith @@ -23,208 +33,104 @@   , newDuplexWith      -- * Classes-  , CanGet (..)+  , CanGet+  , runGet   , read   , isEmpty -  , CanPut (..)+  , CanPut+  , runPut   , write   ) where -import Prelude hiding (read)--import           Control.Concurrent.MVar (MVar, modifyMVar, newMVar)-import qualified Control.Exception as Exception-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           System.Mem.Weak (deRefWeak)+import qualified Data.Binary.IO.Lifted as Lifted+import qualified Data.ByteString as ByteString+import           Prelude hiding (read)+import           System.IO (Handle)  -- * Reader --- | An error that can occur during reading+-- | Alias for 'Lifted.Reader' 'IO' -- -- @since 0.0.1-data ReaderError = ReaderGetError -- ^ Error from the 'Binary.Get' operation-  { readerErrorRemaining :: !ByteString.ByteString-  -- ^ Unconsumed part of the byte stream-  ---  -- @since 0.0.1--  , readerErrorOffset :: !Binary.Get.ByteOffset-  -- ^ Error location represented as an offset into the input-  ---  -- @since 0.0.1--  , readerErrorInput :: !ByteString.ByteString-  -- ^ Input to the 'Binary.Get' operation-  ---  -- @since 0.0.1--  , readerErrorMessage :: !String-  -- ^ Error message-  ---  -- @since 0.0.1-  }-  deriving (Show, Exception.Exception)--newtype StationaryReader = StationaryReader ByteString.ByteString--runStationaryReader-  :: StationaryReader-  -> Binary.Get.Get a-  -> Either ReaderError (StationaryReader, a)-runStationaryReader (StationaryReader stream) getter =-  bimap withError withSuccess (Binary.Get.runGetOrFail getter stream)-  where-    withError (remainingBody, offset, errorMessage) =-      ReaderGetError-        { readerErrorRemaining = remainingBody-        , readerErrorOffset = offset-        , readerErrorInput = stream-        , readerErrorMessage = errorMessage-        }--    withSuccess (tailStream, _, value) = (StationaryReader tailStream, value)--newStationaryReader :: Handle -> IO StationaryReader-newStationaryReader handle = do-  hSetBinaryMode handle True-  StationaryReader <$> ByteString.hGetContents handle--newStationaryReaderWith :: IO ByteString.Strict.ByteString -> IO StationaryReader-newStationaryReaderWith get =-  StationaryReader <$> mkStream get---- | @since 0.0.1-newtype Reader = Reader (MVar StationaryReader)+newtype Reader = Reader+  { unReader :: Lifted.Reader IO } -runReader :: Reader -> Binary.Get a -> IO a-runReader (Reader readerVar) getter =-  modifyMVar readerVar $ \posReader ->-    either Exception.throwIO pure (runStationaryReader posReader getter)+instance Lifted.CanGet Reader IO where+  runGet = runGet . unReader --- | Create a new reader.------ Reading using the 'Reader' may throw 'ReaderError'.------ The internal position of the 'Reader' is not advanced when it throws an exception during reading.--- This has the consequence that if you're trying to read with the same faulty 'Binary.Get'--- operation multiple times, you will always receive an exception.------ Other threads reading from the 'Handle' will interfere with read operations of the 'Reader'.--- However, the 'Reader' itself is thread-safe and can be utilized concurrently.------ Once the 'Handle' reaches EOF, it will be closed.------ The given 'Handle' will be swiched to binary mode via 'hSetBinaryMode'.+-- | Unlifted version of 'Lifted.newReader' -- -- @since 0.0.1 newReader   :: Handle -- ^ Handle that will be read from   -> IO Reader-newReader handle = do-  posReader <- newStationaryReader handle-  Reader <$> newMVar posReader+newReader handle =+  Reader <$> Lifted.newReader handle --- | 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'.+-- | Unlifted version of 'Lifted.newReaderWith'. -- -- @since 0.1.1 newReaderWith-  :: IO ByteString.Strict.ByteString -- ^ Chunk producer+  :: IO ByteString.ByteString -- ^ Chunk producer   -> IO Reader-newReaderWith get = do-  posReader <- newStationaryReaderWith get-  Reader <$> newMVar posReader+newReaderWith get =+  Reader <$> Lifted.newReaderWith get  -- * Writer  -- | @since 0.0.1-newtype Writer = Writer (ByteString.Strict.ByteString -> IO ())+newtype Writer = Writer+  { unWriter :: Lifted.Writer IO } -runWriter :: Writer -> Binary.Put -> IO ()-runWriter (Writer write) putter =-  write (ByteString.toStrict (Binary.Put.runPut putter))+instance Lifted.CanPut Writer IO where+  runPut = runPut . unWriter --- | Create a writer.------ Other threads writing to the same 'Handle' do not interfere with the resulting 'Writer'. The--- 'Writer' may be used concurrently.+-- | Unlifted version of 'Lifted.newWriter' -- -- @since 0.0.1 newWriter   :: Handle -- ^ Handle that will be written to   -> Writer-newWriter handle =-  Writer (ByteString.Strict.hPut handle)+newWriter =+  Writer . Lifted.newWriter --- | Create a writer using a function that handles the output chunks.+-- | Unlifted version of 'Lifted.newWriterWith' -- -- @since 0.1.1 newWriterWith-  :: (ByteString.Strict.ByteString -> IO ()) -- ^ Chunk handler+  :: (ByteString.ByteString -> IO ()) -- ^ Chunk handler   -> Writer newWriterWith =-  Writer+  Writer . Lifted.newWriterWith  -- * Pipe --- | Create a connected pair of 'Reader' and 'Writer'.+-- | Unlifted version of 'Lifted.newPipe' -- -- @since 0.2.0 newPipe :: IO (Reader, Writer)-newPipe = do-  chan <- newIORef mempty-  weakChan <- mkWeakIORef chan (pure ())-  (await, notify) <- newAwaitNotify--  let-    read = do-      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) $ do-        atomicModifyIORef' chan $ \queue ->-          (Deque.snoc msg queue, ())-        runNotify notify--  reader <- newReaderWith read-  let writer = newWriterWith write--  pure (reader, writer)+newPipe = bimap Reader Writer <$> Lifted.newPipe  -- * Duplex --- | Pair of 'Reader' and 'Writer'------ @since 0.0.1+-- | @since 0.0.1 data Duplex = Duplex   { duplexWriter :: !Writer   , duplexReader :: !Reader   } --- | Create a new duplex. The 'Duplex' inherits all the properties of 'Reader' and 'Writer' when--- created with 'newReader' and 'newWriter'.+instance Lifted.CanGet Duplex IO where+  runGet = runGet . duplexReader++instance Lifted.CanPut Duplex IO where+  runPut = runPut . duplexWriter++-- | Unlifted version of 'Lifted.newDuplex' -- -- @since 0.0.1 newDuplex@@ -233,34 +139,35 @@ newDuplex handle =   Duplex (newWriter handle) <$> newReader handle --- | Combines 'newReaderWith' and 'newWriterWith'.+-- | Unlifted version of 'Lifted.newDuplexWith' -- -- @since 0.1.1 newDuplexWith-  :: IO ByteString.Strict.ByteString-  -> (ByteString.Strict.ByteString -> IO ())+  :: IO ByteString.ByteString+  -> (ByteString.ByteString -> IO ())   -> IO Duplex newDuplexWith get push =   Duplex (newWriterWith push) <$> newReaderWith get  -- * Classes --- | @r@ can execute 'Binary.Get' operations+-- | Alias for 'Lifted.CanGet' @r@ 'IO' -- -- @since 0.0.1-class CanGet r where-  runGet-    :: r -- ^ Reader / source-    -> Binary.Get a -- ^ Operation to execute-    -> IO a--instance CanGet Reader where-  runGet = runReader+type CanGet r = Lifted.CanGet r IO -instance CanGet Duplex where-  runGet = runGet . duplexReader+-- | Unlifted version of 'Lifted.runGet'+--+-- @since 0.0.1+runGet+  :: CanGet r+  => r -- ^ Reader / source+  -> Binary.Get a -- ^ Operation to execute+  -> IO a+runGet =+  Lifted.runGet --- | Read something from @r@.+-- | Unlifted version of 'Lifted.read' -- -- @since 0.0.1 read@@ -270,33 +177,28 @@ 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.+-- | Unlifted version of 'Lifted.isEmpty' -- -- @since 0.3.0 isEmpty :: CanGet r => r -> IO Bool isEmpty reader = runGet reader Binary.Get.isEmpty --- | @w@ can execute 'Binary.Put' operations+-- | Alias for 'Lifted.CanPut' @w@ 'IO' -- -- @since 0.0.1-class CanPut w where-  runPut-    :: w -- ^ Writer / target-    -> Binary.Put -- ^ Operation to execute-    -> IO ()--instance CanPut Handle where-  runPut handle putter =-    ByteString.Strict.hPut handle (ByteString.toStrict (Binary.Put.runPut putter))--instance CanPut Writer where-  runPut = runWriter+type CanPut w = Lifted.CanPut w IO -instance CanPut Duplex where-  runPut = runPut . duplexWriter+-- | Unlifted version of 'Lifted.runPut'+--+-- @since 0.0.1+runPut+  :: CanPut w+  => w -- ^ Writer / target+  -> Binary.Put -- ^ Operation to execute+  -> IO ()+runPut = Lifted.runPut --- | Write something to @w@.+-- | Unlifted version of 'Lifted.write' -- -- @since 0.0.1 write@@ -304,22 +206,5 @@   => w -- ^ Write target   -> a -- ^ Value to be written   -> IO ()-write writer value =-  runPut writer (Binary.put value)---- * Utilities---- | Construct a lazy 'ByteString.ByteString' from a function that retrieves chunks.--- Returning an empty chunk indicates the end of the stream.-mkStream :: IO ByteString.Strict.ByteString -> IO ByteString.ByteString-mkStream get =-  readLazily-  where-    read = do-      chunk <- get-      if ByteString.Strict.null chunk then-        pure Empty-      else-        Chunk chunk <$> readLazily--    readLazily = unsafeInterleaveIO read+write =+  Lifted.write

@@ -9,12 +9,18 @@   ) where -import qualified Foreign- import           Data.Word (Word8)+import qualified Foreign import qualified System.IO as IO+import           System.IO.Unsafe (unsafePerformIO) import qualified System.Process as Process +-- | Static pointer that points to a single 'Word8'+someWord8Ptr :: Foreign.Ptr Word8+someWord8Ptr = unsafePerformIO (Foreign.calloc @Word8)++{-# NOINLINE someWord8Ptr #-}+ -- | Await signal from a paired 'Notify'. Returns 'False' if the paired 'Notify' does not exist -- (any more). newtype Await = Await@@ -24,9 +30,9 @@ newtype Notify = Notify   { runNotify :: IO () } +-- | Create a pair of 'Await' and 'Notify. newAwaitNotify :: IO (Await, Notify) newAwaitNotify = do-  buf <- Foreign.calloc @Word8   (read, write) <- Process.createPipe    IO.hSetBuffering read IO.NoBuffering@@ -35,7 +41,7 @@   IO.hSetBinaryMode read True   IO.hSetBinaryMode write True -  let notify = IO.hPutBuf write buf 1-  let await  = (> 0) <$> IO.hGetBufSome read buf 1+  let notify = IO.hPutBuf write someWord8Ptr 1+  let await  = (> 0) <$> IO.hGetBufSome read someWord8Ptr 1    pure (Await await, Notify notify)

@@ -0,0 +1,354 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DerivingVia #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}++-- | Read and write values of types that implement 'Binary.Binary'.+module Data.Binary.IO.Lifted+  ( -- * Reader+    ReaderError (..)++  , Reader (..)+  , newReader+  , newReaderWith+  , mapReader++    -- * Writer+  , Writer (..)+  , newWriter+  , newWriterWith+  , mapWriter++    -- * Pipe+  , newPipe++    -- * Duplex+  , Duplex (..)+  , newDuplex+  , newDuplexWith+  , mapDuplex++    -- * Classes+  , CanGet (..)+  , read+  , isEmpty++  , CanPut (..)+  , write+  )+where++import Prelude hiding (read)++import           Control.Arrow ((&&&))+import qualified Control.Concurrent.Classy as Concurrent+import           Control.Monad (join, unless)+import qualified Control.Monad.Catch as Catch+import           Control.Monad.IO.Class (MonadIO (liftIO))+import           Control.Monad.Trans.Class (MonadTrans (lift))+import           Control.Monad.Trans.Except (ExceptT, except, runExceptT)+import qualified Data.Binary as Binary+import qualified Data.Binary.Get as Get+import           Data.Binary.IO.Internal.AwaitNotify (newAwaitNotify, runAwait, runNotify)+import qualified Data.Binary.Put as Put+import           Data.ByteString (ByteString)+import qualified Data.ByteString as ByteString+import           Data.ByteString.Lazy (toStrict)+import           Data.IORef (atomicModifyIORef', mkWeakIORef, newIORef)+import qualified Deque.Strict as Deque+import           System.IO (Handle, hSetBinaryMode)+import           System.Mem.Weak (deRefWeak)++-- * Reader++-- | An error that can occur during reading+--+-- @since 0.4.0+data ReaderError = ReaderGetError -- ^ Error from the 'Binary.Get' operation+  { readerErrorRemaining :: !ByteString+  -- ^ Unconsumed part of the byte stream+  --+  -- @since 0.4.0++  , readerErrorOffset :: !Get.ByteOffset+  -- ^ Error location represented as an offset into the input+  --+  -- @since 0.4.0++  , readerErrorInput :: !ByteString+  -- ^ Input to the 'Binary.Get' operation+  --+  -- @since 0.4.0++  , readerErrorMessage :: !String+  -- ^ Error message+  --+  -- @since 0.4.0+  }+  deriving stock Show+  deriving anyclass Catch.Exception++newtype StationaryReader m = StationaryReader+  { runStationaryReader+      :: forall a+      .  Binary.Get a+      -> ExceptT ReaderError m (StationaryReader m, a)+  }++newStationaryReaderWith+  :: forall m+  .  Concurrent.MonadConc m+  => m ByteString+  -> m (StationaryReader m)+newStationaryReaderWith getChunk = do+  inputRef <- Concurrent.newIORef ByteString.empty++  let+    make = StationaryReader $ \get -> do+      input <- lift $ Concurrent.readIORef inputRef+      loop $ Get.pushChunk (Get.runGetIncremental get) input++    loop :: Get.Decoder a -> ExceptT ReaderError m (StationaryReader m, a)+    loop = \case+      Get.Fail remainingBody offset errorMessage -> do+        input <- lift $ Concurrent.readIORef inputRef+        except $ Left ReaderGetError+          { readerErrorRemaining = remainingBody+          , readerErrorOffset = offset+          , readerErrorInput = input+          , readerErrorMessage = errorMessage+          }++      Get.Done remainingBody _ value -> Catch.mask_ $ do+        lift $ Concurrent.writeIORef inputRef remainingBody+        pure (make, value)++      Get.Partial continue -> do+        chunk <- lift $ Catch.mask $ \restore -> do+          chunk <- restore getChunk+          if ByteString.null chunk then+            pure Nothing+          else+            Concurrent.atomicModifyIORef' inputRef $ (<> chunk) &&& const (Just chunk)++        loop $ continue chunk++  pure make++-- | @since 0.4.0+newtype Reader m = Reader+  { runReader :: forall a. Binary.Get a -> m a }++-- | Transform the underlying functor.+--+-- @since 0.4.0+mapReader :: (forall a. m a -> n a) -> Reader m -> Reader n+mapReader f (Reader run) = Reader (f . run)++-- | Create a new 'Reader' using an action that provides the chunks.+--+-- The chunk producers indicates the end of the stream by returning an empty+-- 'ByteString.ByteString'.+--+-- Reading using the 'Reader' may throw 'ReaderError'.+--+-- The internal position of the 'Reader' is not advanced when it throws an exception during reading.+-- This has the consequence that if you're trying to read with the same faulty 'Binary.Get'+-- operation multiple times, you will always receive an exception.+--+-- The 'Reader' is safe to use concurrently.+--+-- @since 0.4.0+newReaderWith+  :: Concurrent.MonadConc m+  => m ByteString -- ^ Chunk provider+  -> m (Reader m)+newReaderWith getChunk = do+  posReader <- newStationaryReaderWith getChunk+  mvar <- Concurrent.newMVar posReader+  pure $ Reader $ \get ->+    Concurrent.modifyMVar mvar $ \posReader -> do+      result <- runExceptT $ runStationaryReader posReader get+      either Catch.throwM pure result++-- | Create a new reader.+--+-- Inherits properties from 'newReaderWith'.+--+-- Other threads reading from the 'Handle' will interfere with read operations of the 'Reader'.+-- However, the 'Reader' itself is thread-safe and can be utilized concurrently.+--+-- The given 'Handle' will be swiched to binary mode via 'hSetBinaryMode'.+--+-- @since 0.4.0+newReader+  :: (Concurrent.MonadConc m, MonadIO m)+  => Handle -- ^ Handle to read from+  -> m (Reader m)+newReader handle = do+  liftIO $ hSetBinaryMode handle True+  newReaderWith $ liftIO $ ByteString.hGetSome handle 4096++-- | @r@ can execute 'Binary.Get' operations in @m@+--+-- @since 0.4.0+class CanGet r m where+  runGet :: r -> Binary.Get a -> m a++instance CanGet (Reader m) m where+  runGet = runReader++instance CanGet (Duplex m) m where+  runGet = runGet . duplexReader++-- | Read something from @r@. Inherits properties from 'runGet'.+--+-- @since 0.4.0+read+  :: (CanGet r m, Binary.Binary a)+  => r -- ^ Source to read from+  -> m a+read source = runGet source 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.4.0+isEmpty+  :: CanGet r m+  => r -- ^ Source to check for stream depletion+  -> m Bool+isEmpty source = runGet source Get.isEmpty++-- * Writer++-- | @since 0.4.0+newtype Writer m = Writer+  { runWriter :: Binary.Put -> m () }++-- | Transform the underlying functor.+--+-- @since 0.4.0+mapWriter :: (m () -> n ()) -> Writer m -> Writer n+mapWriter f (Writer write) = Writer (f . write)++-- | Create a writer using a function that handles the output chunks.+--+-- @since 0.4.0+newWriterWith+  :: (ByteString -> m ()) -- ^ Chunk writer+  -> Writer m+newWriterWith write = Writer (write . toStrict . Put.runPut)++-- | Create a writer.+--+-- Other threads writing to the same 'Handle' do not interfere with the resulting 'Writer'. The+-- 'Writer' may be used concurrently.+--+-- @since 0.4.0+newWriter+  :: MonadIO m+  => Handle-- ^ Write target+  -> Writer m+newWriter handle = newWriterWith (liftIO . ByteString.hPut handle)++-- | @w@ can execute 'Binary.Put' operations in @m@+--+-- @since 0.4.0+class CanPut w m where+  runPut :: w -> Binary.Put -> m ()++instance CanPut (Writer m) m where+  runPut = runWriter++instance CanPut (Duplex m) m where+  runPut = runPut . duplexWriter++instance MonadIO m => CanPut Handle m where+  runPut handle = liftIO . ByteString.hPut handle . toStrict . Put.runPut++-- | Write something to @w@.+--+-- @since 0.4.0+write+  :: (CanPut w m, Binary.Binary a)+  => w -- ^ Write target+  -> a -- ^ Value to write+  -> m ()+write sink value = runPut sink $ Binary.put value++-- * Pipe++-- | Create a connected pair of 'Reader' and 'Writer'.+--+-- The 'Reader' will automatically end the stream if the 'Writer' goes out of scope.+--+-- @since 0.4.0+newPipe :: (Concurrent.MonadConc m, MonadIO m) => m (Reader m, Writer m)+newPipe = do+  chan <- liftIO $ newIORef mempty+  weakChan <- liftIO $ mkWeakIORef chan $ pure ()+  (await, notify) <- liftIO newAwaitNotify++  let+    read = do+      mbChan <- deRefWeak weakChan+      case mbChan of+        Nothing -> pure ByteString.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.null msg) $ do+        atomicModifyIORef' chan $ \queue ->+          (Deque.snoc msg queue, ())+        runNotify notify++  reader <- newReaderWith (liftIO read)+  let writer = newWriterWith (liftIO . write)++  pure (reader, writer)++-- * Duplex++-- | Pair of 'Reader' and 'Writer'+--+-- @since 0.4.0+data Duplex m = Duplex+  { duplexWriter :: Writer m+  , duplexReader :: Reader m+  }++-- | Transform the underlying functor.+--+-- @since 0.4.0+mapDuplex :: (forall a. m a -> n a) -> Duplex m -> Duplex n+mapDuplex f (Duplex w r) = Duplex (mapWriter f w) (mapReader f r)++-- | Create a new duplex. The 'Duplex' inherits all the properties of 'Reader' and 'Writer' when+-- created with 'newReader' and 'newWriter'.+--+-- @since 0.4.0+newDuplex+  :: (Concurrent.MonadConc m, MonadIO m)+   => Handle -- ^ Handle to read from and write to+   -> m (Duplex m)+newDuplex handle = Duplex (newWriter handle) <$> newReader handle++-- | Combines 'newReaderWith' and 'newWriterWith'.+--+-- @since 0.4.0+newDuplexWith+  :: Concurrent.MonadConc m+  => m ByteString -- ^ Input chunk producer for 'Reader'+  -> (ByteString -> m ()) -- ^ Chunk writer for 'Writer'+  -> m (Duplex m)+newDuplexWith getChunk writeChunk = Duplex (newWriterWith writeChunk) <$> newReaderWith getChunk

@@ -32,6 +32,6 @@       result <- runAwait await       result `Hspec.shouldBe` False -    Hspec.it "can notify without a paired " $ do+    Hspec.it "can notify without a paired Await" $ do       (_await, notify) <- newAwaitNotify       runNotify notify

@@ -1,12 +1,12 @@ {-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE FlexibleContexts #-}  module Data.Binary.IOSpec (spec) where  import Prelude hiding (read) -import Control.Exception (Exception, throw)-import Control.Monad (join)-+import           Control.Exception (Exception, throw)+import           Control.Monad (join) import           Data.Bifoldable (bitraverse_) import           Data.Binary (Binary (..), encode) import           Data.Binary.IO@@ -15,12 +15,10 @@ import           Data.Foldable (for_) import           Data.List (isInfixOf) import           Data.Typeable (typeOf)--import qualified Test.Hspec as Hspec- import qualified System.IO as IO import           System.IO.Error (ioeGetErrorString, isIllegalOperation) import           System.Process (createPipe)+import qualified Test.Hspec as Hspec  -- | Create a pipe with no buffering on read and write side. createUnbufferedPipe :: IO (IO.Handle, IO.Handle)@@ -191,7 +189,8 @@       write writer "Hello World"       "Hello World" <- read reader -      pure ()+      empty <- isEmpty reader+      empty `Hspec.shouldBe` True      Hspec.it "ends if writer is out of scope" $ do       (reader, _writer) <- newPipe