diff --git a/ChangeLog.md b/ChangeLog.md
--- a/ChangeLog.md
+++ b/ChangeLog.md
@@ -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
diff --git a/binary-io.cabal b/binary-io.cabal
--- a/binary-io.cabal
+++ b/binary-io.cabal
@@ -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
diff --git a/lib/Data/Binary/IO.hs b/lib/Data/Binary/IO.hs
--- a/lib/Data/Binary/IO.hs
+++ b/lib/Data/Binary/IO.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@.
 --
diff --git a/lib/Data/Binary/IO/Internal/AwaitNotify.hs b/lib/Data/Binary/IO/Internal/AwaitNotify.hs
new file mode 100644
--- /dev/null
+++ b/lib/Data/Binary/IO/Internal/AwaitNotify.hs
@@ -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)
diff --git a/test/Data/Binary/IO/Internal/AwaitNotifySpec.hs b/test/Data/Binary/IO/Internal/AwaitNotifySpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Data/Binary/IO/Internal/AwaitNotifySpec.hs
@@ -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
diff --git a/test/Data/Binary/IOSpec.hs b/test/Data/Binary/IOSpec.hs
--- a/test/Data/Binary/IOSpec.hs
+++ b/test/Data/Binary/IOSpec.hs
@@ -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
diff --git a/test/Main.hs b/test/Main.hs
--- a/test/Main.hs
+++ b/test/Main.hs
@@ -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