diff --git a/Data/Conduit/Async.hs b/Data/Conduit/Async.hs
--- a/Data/Conduit/Async.hs
+++ b/Data/Conduit/Async.hs
@@ -10,6 +10,7 @@
 module Data.Conduit.Async ( buffer
                           , ($$&)
                           , gatherFrom
+                          , drainTo
                           ) where
 
 import Control.Applicative
@@ -83,3 +84,31 @@
             Just (Left e)  -> liftIO $ throwIO (e :: SomeException)
             Just (Right r) -> return r
             Nothing        -> gather worker chan
+
+-- | Drain input values into an asynchronous action in the base monad via a
+--   bounded 'TBQueue'.  This is effectively the dual of 'gatherFrom'.
+drainTo :: (MonadIO m, MonadBaseControl IO m)
+        => Int                        -- ^ Size of the queue to create
+        -> (TBQueue (Maybe i) -> m r)  -- ^ Action to consume input values
+        -> Consumer i m r
+drainTo size gather = do
+    chan   <- liftIO $ newTBQueueIO size
+    worker <- lift $ async (gather chan)
+    lift . restoreM =<< scatter worker chan
+  where
+    scatter worker chan = do
+        mval <- await
+        (mx, action) <- liftIO $ atomically $ do
+            mres <- pollSTM worker
+            case mres of
+                Just (Left e)  ->
+                    return (Nothing, liftIO $ throwIO (e :: SomeException))
+                Just (Right r) ->
+                    return (Just r, return ())
+                Nothing        -> do
+                    writeTBQueue chan mval
+                    return (Nothing, return ())
+        action
+        case mx of
+            Just x  -> return x
+            Nothing -> scatter worker chan
diff --git a/stm-conduit.cabal b/stm-conduit.cabal
--- a/stm-conduit.cabal
+++ b/stm-conduit.cabal
@@ -1,5 +1,5 @@
 Name:                stm-conduit
-Version:             2.1.3
+Version:             2.1.4
 Synopsis:            Introduces conduits to channels, and promotes using
                      conduits concurrently.
 Description:         Provides two simple conduit wrappers around STM
diff --git a/test/Test.hs b/test/Test.hs
--- a/test/Test.hs
+++ b/test/Test.hs
@@ -32,6 +32,7 @@
         testGroup "Async functions" [
                   testCase "buffer" test_buffer
                 , testCase "gatherFrom" test_gatherFrom
+                , testCase "drainTo" test_drainTo
             ],
         testGroup "Bug fixes" [
                   testCase "multipleWriters" test_multipleWriters
@@ -91,3 +92,13 @@
         f q x = do
             atomically $ writeTBQueue q x
             return q
+
+test_drainTo = do
+    sum' <- CL.sourceList [1..100] $$ drainTo 128 (go 0)
+    assertEqual "sum computed using drainTo" sum' 5050
+  where
+    go acc queue = do
+        mres <- atomically $ readTBQueue queue
+        case mres of
+            Nothing  -> return acc
+            Just res -> go (acc + res) queue