diff --git a/Data/Conduit/Async.hs b/Data/Conduit/Async.hs
--- a/Data/Conduit/Async.hs
+++ b/Data/Conduit/Async.hs
@@ -29,12 +29,16 @@
 buffer size input output = do
     chan <- liftIO $ newTBQueueIO size
     control $ \runInIO ->
-        withAsync (runInIO $ input $$ mapM_ (send chan)) $ \input' ->
+        withAsync (runInIO $ sender chan) $ \input' ->
             withAsync (runInIO $ recv chan $$ output) $ \output' -> do
                 link2 input' output'
                 wait output'
   where
-    send chan = liftIO . atomically . writeTBQueue chan . Just
+    send chan = liftIO . atomically . writeTBQueue chan
+
+    sender chan = do
+        input $$ mapM_ (send chan . Just)
+        send chan Nothing
 
     recv chan = do
         mx <- liftIO $ atomically $ readTBQueue 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.1
+Version:             2.1.2
 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
@@ -16,6 +16,7 @@
 import Control.Concurrent.STM.TMQueue
 import Data.Conduit
 import Data.Conduit.List as CL
+import Data.Conduit.Async
 import Data.Conduit.TMChan
 import Data.Conduit.TQueue
 
@@ -28,7 +29,8 @@
                 , testCase "simpleList using TMQueue" test_simpleMQueue
             ],
         testGroup "Bug fixes" [
-                testCase "multipleWriters" test_multipleWriters
+                  testCase "multipleWriters" test_multipleWriters
+                , testCase "asyncOperator" test_asyncOperator
             ]
     ]
 
@@ -62,3 +64,11 @@
                                                             ] 3
                           xs <- runResourceT $ ms $$ consume
                           liftIO $ assertEqual "for the numbers [1..10] and [11..20]," [1..20] $ sort xs
+
+test_asyncOperator = do sum'  <- CL.sourceList [1..n] $$ CL.fold (+) 0
+                        assertEqual ("for the sum of 1 to " ++ show n) sum sum'
+                        sum'' <- CL.sourceList [1..n] $$& CL.fold (+) 0
+                        assertEqual "for the sum computed with the $$ and the $$&" sum' sum''
+    where
+        n = 100
+        sum = n * (n+1) / 2