diff --git a/Data/Conduit/TMChan.hs b/Data/Conduit/TMChan.hs
--- a/Data/Conduit/TMChan.hs
+++ b/Data/Conduit/TMChan.hs
@@ -50,6 +50,8 @@
                            , mergeSources
                            ) where
 
+import Control.Applicative
+import Control.Monad
 import Control.Monad.IO.Class ( liftIO )
 import Control.Monad.Trans.Resource
 import Control.Concurrent.STM
@@ -67,7 +69,7 @@
 chanSource ch reader closer = src
     where
         src = Source pull close
-        pull = do a <- liftIO . atomically $ reader ch
+        pull = do a <- liftSTM $ reader ch
                   case a of
                     Just x  -> return $ Open src x
                     Nothing -> return Closed
@@ -83,7 +85,7 @@
 chanSink ch writer closer = sink
     where
         sink = SinkData push close
-        push input = do liftIO . atomically $ writer ch input
+        push input = do liftSTM $ writer ch input
                         return $ Processing push close
         close = liftIO . atomically $ closer ch
 {-# INLINE chanSink #-}
@@ -121,6 +123,15 @@
 
 infixl 5 >=<
 
+-- | Modifies a TVar, returning its new value.
+modifyTVar'' :: TVar a -> (a -> a) -> STM a
+modifyTVar'' tv f = do x <- f <$> readTVar tv
+                       writeTVar tv x
+                       return x
+
+liftSTM :: ResourceIO m => STM a -> ResourceT m a
+liftSTM = liftIO . atomically
+
 -- | Combines two sources with an unbounded channel, creating a new source
 --   which pulls data from a mix of the two sources: whichever produces first.
 --
@@ -130,11 +141,14 @@
       => Source m a
       -> Source m a
       -> ResourceT m (Source m a)
-sa >=< sb = do c <- liftIO . atomically $ newTMChan 
-               _ <- resourceForkIO $ sa $$ sinkTMChan c
-               _ <- resourceForkIO $ sb $$ sinkTMChan c
-               return $ sourceTMChan c
+sa >=< sb = mergeSources [ sa, sb ] 16
+{-# INLINE (>=<) #-}
 
+decRefcount :: TVar Int -> TBMChan a ->  STM ()
+decRefcount tv chan = do n <- modifyTVar'' tv (subtract 1)
+                         when (n == 0) $
+                            closeTBMChan chan
+
 -- | Merges a list of sources, putting them all into a bounded channel, and
 --   returns a source which can be pulled from to pull from all the given
 --   sources in a first-come-first-serve basis.
@@ -146,5 +160,6 @@
              -> Int -- ^ The bound of the intermediate channel.
              -> ResourceT m (Source m a)
 mergeSources sx bound = do c <- liftIO . atomically $ newTBMChan bound
-                           mapM_ (\s -> resourceForkIO $ s $$ sinkTBMChan c) sx
+                           refcount <- liftSTM . newTVar $ length sx
+                           mapM_ (\s -> resourceForkIO $ s $$ chanSink c writeTBMChan $ decRefcount refcount) sx
                            return $ sourceTBMChan c
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:             0.2.4
+Version:             0.2.4.1
 Synopsis:            Introduces conduits to channels, and promotes using
                      conduits concurrently.
 Description:         Provides two simple conduit wrappers around STM
@@ -32,7 +32,7 @@
     main-is:        Test.hs
     hs-source-dirs: test/
 
-    ghc-options:    -rtsopts=all
+    ghc-options:    -rtsopts=all -threaded
 
     Build-Depends:  base ==4.*,
                     QuickCheck >= 2,
diff --git a/test/Test.hs b/test/Test.hs
new file mode 100644
--- /dev/null
+++ b/test/Test.hs
@@ -0,0 +1,42 @@
+module Main ( main ) where
+
+import Data.List
+
+import Test.Framework (defaultMain, testGroup)
+import Test.Framework.Providers.HUnit
+import Test.Framework.Providers.QuickCheck2 (testProperty)
+
+import Test.QuickCheck
+import Test.HUnit
+
+import Control.Concurrent
+import Control.Concurrent.STM
+import Data.Conduit
+import Data.Conduit.List
+import Data.Conduit.TMChan
+
+main = defaultMain tests
+
+tests = [
+        testGroup "Behaves to spec" [
+                testCase "simpleList" test_simpleList
+            ],
+        testGroup "Bug fixes" [
+                testCase "multipleWriters" test_multipleWriters
+            ]
+    ]
+
+test_simpleList = do chan <- atomically $ newTMChan
+                     forkIO . runResourceT $ sourceList testList $$ sinkTMChan chan
+                     lst' <- runResourceT $ sourceTMChan chan $$ consume
+                     assertEqual "for the numbers [1..10000]," testList lst'
+                     closed <- atomically $ isClosedTMChan chan
+                     assertBool "channel is closed after running" closed
+    where
+        testList = [1..10000]
+
+test_multipleWriters = do ms <- runResourceT $ mergeSources [ sourceList [1..10]
+                                                           , sourceList [11..20]
+                                                           ] 3
+                          xs <- runResourceT $ ms $$ consume
+                          assertEqual "for the numbers [1..10] and [11..20]," [1..20] $ sort xs