diff --git a/Changelog.md b/Changelog.md
--- a/Changelog.md
+++ b/Changelog.md
@@ -1,5 +1,13 @@
 # Changelog for net-mqtt
 
+## 0.6.2.0
+
+Added low-level support for correlated responses.
+
+I'm not completely sure how generally useful this interface is, but
+I've been using it in a client that's implementing an RPC type
+interface across MQTT for a bit now.
+
 ## 0.6.1.1
 
 Add connection checks to publish phases.
diff --git a/net-mqtt.cabal b/net-mqtt.cabal
--- a/net-mqtt.cabal
+++ b/net-mqtt.cabal
@@ -4,10 +4,10 @@
 --
 -- see: https://github.com/sol/hpack
 --
--- hash: 6c69102312ff4cecc87356752a2c3125e4e9aa010ece324b5d191f234c89616c
+-- hash: 5602a18bc959d07604ceba74369d151f6f3296000f0673b64a4539234b49d3e5
 
 name:           net-mqtt
-version:        0.6.1.1
+version:        0.6.2.0
 synopsis:       An MQTT Protocol Implementation.
 description:    Please see the README on GitHub at <https://github.com/dustin/mqtt-hs#readme>
 category:       Network
diff --git a/src/Network/MQTT/Client.hs b/src/Network/MQTT/Client.hs
--- a/src/Network/MQTT/Client.hs
+++ b/src/Network/MQTT/Client.hs
@@ -15,6 +15,7 @@
 are supported over plain TCP, TLS, WebSockets and Secure WebSockets.
 -}
 
+{-# LANGUAGE NamedFieldPuns    #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE RecordWildCards   #-}
 
@@ -30,7 +31,8 @@
   subscribe, unsubscribe, publish, publishq, pubAliased,
   svrProps, MQTTException(..),
   -- * Low-level bits
-  runMQTTConduit, MQTTConduit, isConnectedSTM
+  runMQTTConduit, MQTTConduit, isConnectedSTM,
+  registerCorrelated, unregisterCorrelated
   ) where
 
 import           Control.Concurrent         (myThreadId, threadDelay)
@@ -108,6 +110,7 @@
   , _outA     :: TVar (Map Topic Word16)
   , _inA      :: TVar (Map Word16 Topic)
   , _svrProps :: TVar [Property]
+  , _corr     :: TVar (Map BL.ByteString MessageCallback)
   }
 
 -- | Configuration for setting up an MQTT client.
@@ -287,6 +290,7 @@
   _outA <- newTVarIO mempty
   _inA <- newTVarIO mempty
   _svrProps <- newTVarIO mempty
+  _corr <- newTVarIO mempty
   let _cb = _msgCB
       cli = MQTTClient{..}
 
@@ -433,9 +437,14 @@
           | otherwise = notify
 
           where
+            cdata = foldr f Nothing _pubProps
+              where f (PropCorrelationData x) _ = Just x
+                    f _ o                       = o
+
             notify = do
               topic <- resolveTopic (foldr aliasID Nothing _pubProps)
-              E.evaluate . force =<< case _cb of
+              corrs <- readTVarIO _corr
+              E.evaluate . force =<< case maybe _cb (\cd -> Map.findWithDefault _cb cd corrs) cdata of
                                        NoCallback         -> pure ()
                                        SimpleCallback f   -> f c topic _pubBody _pubProps
                                        LowLevelCallback f -> f c pr{_pubTopic=textToBL topic}
@@ -672,3 +681,13 @@
           v = fromMaybe (if n > mv then 0 else n) cur
       when (v > 0) $ writeTVar _outA (Map.insert t v as)
       pure (maybe t (const "") cur, v)
+
+-- | Register a callback handler for a message with the given correlated data identifier.
+--
+-- This registration will remain in place until unregisterCorrelated is called to remove it.
+registerCorrelated :: MQTTClient -> BL.ByteString -> MessageCallback -> STM ()
+registerCorrelated MQTTClient{_corr} bs cb = modifyTVar' _corr (Map.insert bs cb)
+
+-- | Unregister a callback handler for the given correlated data identifier.
+unregisterCorrelated :: MQTTClient -> BL.ByteString -> STM ()
+unregisterCorrelated MQTTClient{_corr} bs = modifyTVar' _corr (Map.delete bs)