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