diff --git a/Data/Conduit/Async.hs b/Data/Conduit/Async.hs
--- a/Data/Conduit/Async.hs
+++ b/Data/Conduit/Async.hs
@@ -15,39 +15,33 @@
                           , drainTo
                           ) where
 
-#if __GLASGOW_HASKELL__ < 710
-import Control.Applicative
-#endif
-import Control.Concurrent.Async.Lifted
-import Control.Concurrent.STM
-import Control.Exception.Lifted
 import Control.Monad.IO.Class
 import Control.Monad.Loops
 import Control.Monad.Trans.Class
-import Control.Monad.Trans.Control
 import Data.Conduit
-
 import Data.Conduit.Async.Composition
+import Data.Foldable
+import UnliftIO
 
 -- | Gather output values asynchronously from an action in the base monad and
 --   then yield them downstream.  This provides a means of working around the
 --   restriction that 'ConduitM' cannot be an instance of 'MonadBaseControl'
 --   in order to, for example, yield values from within a Haskell callback
 --   function called from a C library.
-gatherFrom :: (MonadIO m, MonadBaseControl IO m)
+gatherFrom :: (MonadIO m, MonadUnliftIO m)
            => Int                -- ^ Size of the queue to create
            -> (TBQueue o -> m ()) -- ^ Action that generates output values
-           -> Producer m o
+           -> ConduitT () o m ()
 gatherFrom size scatter = do
     chan   <- liftIO $ newTBQueueIO size
     worker <- lift $ async (scatter chan)
-    lift . restoreM =<< gather worker chan
+    gather worker chan
   where
     gather worker chan = do
         (xs, mres) <- liftIO $ atomically $ do
             xs <- whileM (not <$> isEmptyTBQueue chan) (readTBQueue chan)
             (xs,) <$> pollSTM worker
-        Prelude.mapM_ yield xs
+        traverse_ yield xs
         case mres of
             Just (Left e)  -> liftIO $ throwIO (e :: SomeException)
             Just (Right r) -> return r
@@ -55,14 +49,14 @@
 
 -- | 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)
+drainTo :: (MonadIO m, MonadUnliftIO m)
         => Int                        -- ^ Size of the queue to create
         -> (TBQueue (Maybe i) -> m r)  -- ^ Action to consume input values
-        -> Consumer i m r
+        -> ConduitT i Void m r
 drainTo size gather = do
     chan   <- liftIO $ newTBQueueIO size
     worker <- lift $ async (gather chan)
-    lift . restoreM =<< scatter worker chan
+    scatter worker chan
   where
     scatter worker chan = do
         mval <- await
diff --git a/Data/Conduit/Async/Composition.hs b/Data/Conduit/Async/Composition.hs
--- a/Data/Conduit/Async/Composition.hs
+++ b/Data/Conduit/Async/Composition.hs
@@ -24,14 +24,8 @@
                                       , runCConduit
                                       ) where
 
-#if __GLASGOW_HASKELL__ < 710
-import           Control.Applicative
-#endif
 import Conduit
-import qualified "async" Control.Concurrent.Async as A
-import Control.Concurrent.Async.Lifted hiding (link2)
-import Control.Concurrent.STM
-import Control.Exception (finally)
+import Control.Concurrent.STM (orElse, check)
 import Control.Monad hiding (forM_)
 import Control.Monad.Loops
 import Control.Monad.Trans.Resource
@@ -40,13 +34,10 @@
 import qualified Data.Conduit.List as CL
 import Data.Foldable (forM_)
 import Data.Serialize
-import Data.Void
-#if __GLASGOW_HASKELL__ > 710
 import GHC.Exts (Constraint)
-#endif
-import GHC.Prim
 import System.Directory (removeFile)
-import System.IO
+import System.IO (openBinaryTempFile)
+import UnliftIO
 
 -- | Concurrently join the producer and consumer, using a bounded queue of the
 -- given size. The producer will block when the queue is full, if it is
@@ -161,7 +152,7 @@
 --
 -- >>> runResourceT $ bufferToFile 1 Nothing "/tmp" (CL.sourceList [1,2,3]) CL.consume
 -- [1,2,3]
-bufferToFile :: (CFConduitLike c1, CFConduitLike c2, Serialize x, MonadBaseControl IO m, MonadIO m, MonadResource m)
+bufferToFile :: (CFConduitLike c1, CFConduitLike c2, Serialize x, MonadUnliftIO m, MonadResource m, MonadThrow m)
                 => Int -- ^ Size of the bounded queue in memory
                 -> Maybe Int -- ^ Max elements to keep on disk at one time
                 -> FilePath -- ^ Directory to write temp files to
@@ -212,20 +203,20 @@
   -- equivalent of 'runConduit'.
   --
   -- The underlying monad must always be an instance of
-  -- 'MonadBaseControl IO'.  If the conduits is a 'CFConduit', it must
+  -- 'MonadUnliftIO'.  If the conduits is a 'CFConduit', it must
   -- additionally be a in instance of 'MonadResource'.
   runCConduit :: (RunConstraints c m) => c () Void m r -> m r
 
-instance CCatable ConduitM ConduitM CConduit where
+instance CCatable ConduitT ConduitT CConduit where
   buffer' i a b = buffer' i (Single a) (Single b)
 
-instance CCatable ConduitM CConduit CConduit where
+instance CCatable ConduitT CConduit CConduit where
   buffer' i a b = buffer' i (Single a) b
 
-instance CCatable ConduitM CFConduit CFConduit where
+instance CCatable ConduitT CFConduit CFConduit where
   buffer' i a b = buffer' i (asCFConduit a) b
 
-instance CCatable CConduit ConduitM CConduit where
+instance CCatable CConduit ConduitT CConduit where
   buffer' i a b = buffer' i a (Single b)
 
 instance CCatable CConduit CConduit CConduit where
@@ -235,7 +226,7 @@
 instance CCatable CConduit CFConduit CFConduit where
   buffer' i a b = buffer' i (asCFConduit a) b
 
-instance CCatable CFConduit ConduitM CFConduit where
+instance CCatable CFConduit ConduitT CFConduit where
   buffer' i a b = buffer' i a (asCFConduit b)
 
 instance CCatable CFConduit CConduit CFConduit where
@@ -246,12 +237,12 @@
   buffer' i (FMultiple i' a as) b = FMultiple i' a (buffer' i as b)
   buffer' i (FMultipleF bufsz dsksz tmpDir a as) b = FMultipleF bufsz dsksz tmpDir a (buffer' i as b)
 
-instance CRunnable ConduitM where
-  type RunConstraints ConduitM m = (Monad m)
+instance CRunnable ConduitT where
+  type RunConstraints ConduitT m = (MonadUnliftIO m, Monad m)
   runCConduit = runConduit
 
 instance CRunnable CConduit where
-  type RunConstraints CConduit m = (MonadBaseControl IO m, MonadIO m)
+  type RunConstraints CConduit m = (MonadUnliftIO m, MonadIO m)
   runCConduit (Single c) = runConduit c
   runCConduit (Multiple bufsz c cs) = do
     chan <- liftIO $ newTBQueueIO bufsz
@@ -259,7 +250,7 @@
       stage chan c' cs
 
 instance CRunnable CFConduit where
-  type RunConstraints CFConduit m = (MonadBaseControl IO m, MonadIO m, MonadResource m)
+  type RunConstraints CFConduit m = (MonadThrow m, MonadUnliftIO m, MonadIO m, MonadResource m)
   runCConduit (FSingle c) = runConduit c
   runCConduit (FMultiple bufsz c cs) = do
     chan <- liftIO $ newTBQueueIO bufsz
@@ -277,26 +268,22 @@
 -- | A "concurrent conduit", in which the stages run in parallel with
 -- a buffering queue between them.
 data CConduit i o m r where
-  Single :: ConduitM i o m r -> CConduit i o m r
-  Multiple :: Int -> ConduitM i x m () -> CConduit x o m r -> CConduit i o m r
-
--- C.C.A.L's link2 has the wrong type:  https://github.com/maoe/lifted-async/issues/16
-link2 :: MonadBase IO m => Async a -> Async b -> m ()
-link2 = (liftBase .) . A.link2
+  Single :: ConduitT i o m r -> CConduit i o m r
+  Multiple :: Int -> ConduitT i x m () -> CConduit x o m r -> CConduit i o m r
 
 -- Combines a producer with a queue, sending it everything the
 -- producer produces.
-sender :: (MonadIO m) => TBQueue (Maybe o) -> ConduitM () o m () -> m ()
+sender :: (MonadIO m) => TBQueue (Maybe o) -> ConduitT () o m () -> m ()
 sender chan input = do
-  input $$ mapM_C (send chan . Just)
+  runConduit $ input .| mapM_C (send chan . Just)
   send chan Nothing
 
 -- One "layer" of withAsync in a CConduit run.
-stage :: (MonadBaseControl IO m, MonadIO m) => TBQueue (Maybe i) -> Async x -> CConduit i Void m r -> m r
+stage :: (MonadUnliftIO m) => TBQueue (Maybe i) -> Async x -> CConduit i Void m r -> m r
 stage chan prevAsync (Single c) =
   -- The last layer; feed the output of "chan" into the conduit and
   -- wait for the result.
-  withAsync (receiver chan $$ c) $ \c' -> do
+  withAsync (runConduit $ receiver chan .| c) $ \c' -> do
     link2 prevAsync c'
     wait c'
 stage chan prevAsync (Multiple bufsz c cs) = do
@@ -304,13 +291,13 @@
   -- layer's conduit process it, and send the conduit's output to the
   -- next layer.
   chan' <- liftIO $ newTBQueueIO bufsz
-  withAsync (sender chan' $ receiver chan =$= c) $ \c' -> do
+  withAsync (sender chan' $ receiver chan .| c) $ \c' -> do
     link2 prevAsync c'
     stage chan' c' cs
 
 -- A Producer which produces the values of the given channel until
 -- Nothing is received.  This is the other half of "sender".
-receiver :: (MonadIO m) => TBQueue (Maybe o) -> ConduitM () o m ()
+receiver :: (MonadIO m) => TBQueue (Maybe o) -> ConduitT () o m ()
 receiver chan = do
   mx <- recv chan
   case mx of
@@ -320,14 +307,14 @@
 -- | A "concurrent conduit", in which the stages run in parallel with
 -- a buffering queue and possibly a disk file between them.
 data CFConduit i o m r where
-  FSingle :: ConduitM i o m r -> CFConduit i o m r
-  FMultiple :: Int -> ConduitM i x m () -> CFConduit x o m r -> CFConduit i o m r
-  FMultipleF :: (Serialize x) => Int -> Maybe Int -> FilePath -> ConduitM i x m () -> CFConduit x o m r -> CFConduit i o m r
+  FSingle :: ConduitT i o m r -> CFConduit i o m r
+  FMultiple :: Int -> ConduitT i x m () -> CFConduit x o m r -> CFConduit i o m r
+  FMultipleF :: (Serialize x) => Int -> Maybe Int -> FilePath -> ConduitT i x m () -> CFConduit x o m r -> CFConduit i o m r
 
 class CFConduitLike a where
   asCFConduit :: a i o m r -> CFConduit i o m r
 
-instance CFConduitLike ConduitM where
+instance CFConduitLike ConduitT where
   asCFConduit = FSingle
 
 instance CFConduitLike CConduit where
@@ -338,7 +325,7 @@
   asCFConduit = id
 
 data BufferContext m a = BufferContext { chan :: TBQueue a
-                                       , restore :: TQueue (Source m a)
+                                       , restore :: TQueue (ConduitT () a m ())
                                        , slotsFree :: TVar (Maybe Int)
                                        , done :: TVar Bool
                                        , tempDir :: FilePath
@@ -347,22 +334,22 @@
 -- The file-backed equivlent of "sender".  This sends the values
 -- generated by "input" to the "chan" in the BufferContext until it
 -- gets full, then flushes it to disk via "persistChan".
-fsender :: (MonadIO m, MonadResource m, Serialize x) => BufferContext m x -> ConduitM () x m () -> m ()
+fsender :: (MonadThrow m, MonadResource m, Serialize x) => BufferContext m x -> ConduitT () x m () -> m ()
 fsender bc@BufferContext{..} input = do
-  input $$ mapM_C $ \x -> join $ liftIO $ atomically $ do
+  runConduit $ input .| (mapM_C $ \x -> join $ liftIO $ atomically $ do
     (writeTBQueue chan x >> return (return ())) `orElse` do
       action <- persistChan bc
       writeTBQueue chan x
-      return action
+      return action)
   liftIO $ atomically $ writeTVar done True
 
 -- Connect a stage to another stage via either an in-memory queue or a
 -- disk buffer.  This is the file-backed equivalent of "stage".
-fstage :: (MonadBaseControl IO m, MonadIO m, MonadResource m) => ConduitM () i m () -> Async x -> CFConduit i Void m r -> m r
+fstage :: (MonadThrow m, MonadUnliftIO m, MonadResource m) => ConduitT () i m () -> Async x -> CFConduit i Void m r -> m r
 fstage prevStage prevAsync (FSingle c) =
   -- The final conduit in the chain; just accept everything from
   -- the previous stage and wait for the result.
-  withAsync (prevStage $$ c) $ \c' -> do
+  withAsync (runConduit $ prevStage .| c) $ \c' -> do
     link2 prevAsync c'
     wait c'
 fstage prevStage prevAsync (FMultiple bufsz c cs) = do
@@ -370,7 +357,7 @@
   -- channel, so it just uses "sender" and "reciever" in the same way
   -- "stage" does.
   chan' <- liftIO $ newTBQueueIO bufsz
-  withAsync (sender chan' $ prevStage =$= c) $ \c' -> do
+  withAsync (sender chan' $ prevStage .| c) $ \c' -> do
     link2 prevAsync c'
     fstage (receiver chan') c' cs
 fstage prevStage prevAsync (FMultipleF bufsz dsksz tempDir c cs) = do
@@ -381,13 +368,13 @@
                                <*> newTVarIO dsksz
                                <*> newTVarIO False
                                <*> pure tempDir
-  withAsync (fsender bc $ prevStage =$= c) $ \c' -> do
+  withAsync (fsender bc $ prevStage .| c) $ \c' -> do
     link2 prevAsync c'
     fstage (freceiver bc) c' cs
 
 -- Receives from disk files or the in-memory queue if no spill-to-disk
 -- has occurred.
-freceiver :: (MonadIO m) => BufferContext m o -> ConduitM () o m ()
+freceiver :: (MonadIO m) => BufferContext m o -> ConduitT () o m ()
 freceiver BufferContext{..} = loop where
   loop = do
     (src, exit) <- liftIO $ atomically $ do
@@ -400,7 +387,7 @@
 
 -- The channel is full, so (return an action which will) spill it to disk, unless too
 -- many items are there already.
-persistChan :: (MonadIO m, MonadResource m, Serialize o) => BufferContext m o -> STM (m ())
+persistChan :: (MonadThrow m, MonadResource m, Serialize o) => BufferContext m o -> STM (m ())
 persistChan BufferContext{..} = do
   xs <- exhaust chan
   mslots <- readTVar slotsFree
@@ -409,8 +396,8 @@
   filePath <- newEmptyTMVar
   writeTQueue restore $ do
     (path, key) <- liftIO $ atomically $ takeTMVar filePath
-    CB.sourceFile path $= do
-      C.conduitGet get
+    CB.sourceFile path .| do
+      C.conduitGet2 get
       liftIO $ atomically $ modifyTVar slotsFree (fmap (+ len))
       release key
   case xs of
@@ -420,7 +407,7 @@
      return $ do
        (key, (path, h)) <- allocate (openBinaryTempFile tempDir "conduit.bin") (\(path, h) -> hClose h `finally` removeFile path)
        liftIO $ do
-         CL.sourceList xs $= C.conduitPut put $$ CB.sinkHandle h
+         runConduit $ CL.sourceList xs .| C.conduitPut put .| CB.sinkHandle h
          hClose h
          atomically $ putTMVar filePath (path, key)
 
diff --git a/Data/Conduit/TMChan.hs b/Data/Conduit/TMChan.hs
--- a/Data/Conduit/TMChan.hs
+++ b/Data/Conduit/TMChan.hs
@@ -1,4 +1,4 @@
-{-# LANGUAGE CPP #-}
+{-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE NoMonomorphismRestriction #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE RankNTypes #-}
@@ -18,8 +18,14 @@
 --   data (pictures, in this case) will be streamed down the channel to whatever
 --   is on the other side.
 --
---   >    _ <- forkIO . runResourceT $ loadTextures lotsOfPictures $$ sinkTBMChan chan
+--   >    _ <- forkIO . runResourceT $ do
+--   >          _ <- register $ atomically $ closeTBMChan chan
+--   >          loadTextures lotsOfPictures $$ sinkTBMChan chan
 --
+--   We register closing function explicitly, because starting with version
+--   @1.3.0@ @conduits@ library no longer maintain resources, so this is the
+--   only way to safely close channel in case of exceptions.
+--
 --   Finally, we connect something to the other end of the channel. In this
 --   case, we connect a sink which uploads the textures one by one to the
 --   graphics card.
@@ -58,49 +64,48 @@
                            , mergeConduits
                            ) where
 
-#if __GLASGOW_HASKELL__ < 710
-import           Control.Applicative
-#endif
 import Control.Monad
 import Control.Monad.IO.Class ( liftIO, MonadIO )
 import Control.Monad.Trans.Class
 import Control.Monad.Trans.Resource
-import Control.Concurrent (killThread)
-import Control.Concurrent.STM
+import Control.Concurrent (killThread, forkIOWithUnmask)
+import Control.Concurrent.STM hiding (atomically)
 import Control.Concurrent.STM.TBMChan
 import Control.Concurrent.STM.TMChan
 
-import qualified Control.Exception.Lifted as Lifted
-
 import Data.Conduit
+import Data.Foldable
 import qualified Data.Conduit.List as CL
+import UnliftIO as Lifted
 
+-- | Convert channel into the source.
+--
+-- *N.B* Since version 4.0 this function does not close the
+-- channel if downstream is closed.
 chanSource
     :: MonadIO m
     => chan                     -- ^ The channel.
     -> (chan -> STM (Maybe a))  -- ^ The 'read' function.
-    -> (chan -> IO  ())         -- ^ The 'close' function.
-    -> Source m a
-chanSource ch reader closer =
+    -> ConduitT z a m ()
+chanSource ch reader =
     loop
   where
     loop = do
         a <- liftSTM $ reader ch
         case a of
-            Just x  -> yieldOr x close >> loop
+            Just x  -> yield x >> loop
             Nothing -> return ()
-    close = liftIO $ closer ch
 {-# INLINE chanSource #-}
 
+-- | Convert channel into the consumer.
+--
+-- *N.B*
 chanSink
     :: MonadIO m
     => chan                     -- ^ The channel.
     -> (chan -> a -> STM ())    -- ^ The 'write' function.
-    -> (chan -> STM ())         -- ^ The 'close' function.
-    -> Sink a m ()
-chanSink ch writer closer = do
-    CL.mapM_ $ liftIO . atomically . writer ch
-    liftSTM $ closer ch
+    -> ConduitT a z m ()
+chanSink ch writer = CL.mapM_ $ liftIO . atomically . writer ch
 {-# INLINE chanSink #-}
 
 -- | A simple wrapper around a TBMChan. As data is pushed into the channel, the
@@ -108,15 +113,15 @@
 --   channel is closed, the source will close also.
 --
 --   If the channel fills up, the pipeline will stall until values are read.
-sourceTBMChan :: MonadIO m => TBMChan a -> Source m a
-sourceTBMChan ch = chanSource ch readTBMChan (atomically . closeTBMChan)
+sourceTBMChan :: MonadIO m => TBMChan a -> ConduitT () a m ()
+sourceTBMChan ch = chanSource ch readTBMChan
 {-# INLINE sourceTBMChan #-}
 
 -- | A simple wrapper around a TMChan. As data is pushed into the channel, the
 --   source will read it and pass it down the conduit pipeline. When the
 --   channel is closed, the source will close also.
-sourceTMChan :: MonadIO m => TMChan a -> Source m a
-sourceTMChan ch = chanSource ch readTMChan (atomically . closeTMChan)
+sourceTMChan :: MonadIO m => TMChan a -> ConduitT () a m ()
+sourceTMChan ch = chanSource ch readTMChan
 {-# INLINE sourceTMChan #-}
 
 -- | A simple wrapper around a TBMChan. As data is pushed into the sink, it
@@ -124,27 +129,26 @@
 --   the sink will block until space frees up.
 sinkTBMChan :: MonadIO m
             => TBMChan a
-            -> Bool -- ^ Should the channel be closed when the sink is closed?
-            -> Sink a m ()
-sinkTBMChan ch close = chanSink ch writeTBMChan (when close . closeTBMChan)
+            -> ConduitT a z m ()
+sinkTBMChan ch = chanSink ch writeTBMChan
 {-# INLINE sinkTBMChan #-}
 
 -- | A simple wrapper around a TMChan. As data is pushed into this sink, it
 --   will magically begin to appear in the channel.
 sinkTMChan :: MonadIO m
            => TMChan a
-           -> Bool -- ^ Should the channel be closed when the sink is closed?
-           -> Sink a m ()
-sinkTMChan ch close = chanSink ch writeTMChan (when close . closeTMChan)
+           -> ConduitT a z m ()
+sinkTMChan ch = chanSink ch writeTMChan
 {-# INLINE sinkTMChan #-}
 
 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
+modifyTVar'' tv f = do
+  !x <- f <$> readTVar tv
+  writeTVar tv x
+  return x
 
 liftSTM :: forall (m :: * -> *) a. MonadIO m => STM a -> m a
 liftSTM = liftIO . atomically
@@ -154,17 +158,18 @@
 --
 --   The order of the new source's data is undefined, but it will be some
 --   combination of the two given sources.
-(>=<) :: (MonadResource mi, MonadIO mo, MonadBaseControl IO mi)
-      => Source mi a
-      -> Source mi a
-      -> mi (Source mo a)
+(>=<) :: (MonadResource mi, MonadIO mo, MonadUnliftIO mi)
+      => ConduitT () a mi ()
+      -> ConduitT () a mi ()
+      -> mo (ConduitT () a mi ())
 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
+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
@@ -182,29 +187,35 @@
 --   @before 3.0
 --   Spawned threads are not guaranteed to be closed. This may happen if
 --   Source was closed before all it's input were closed.
-mergeSources :: (MonadResource mi, MonadIO mo, MonadBaseControl IO mi)
-             => [Source mi a] -- ^ The sources to merge.
+mergeSources :: (MonadResource mi, MonadIO mo, MonadUnliftIO mi)
+             => [ConduitT () a mi ()] -- ^ The sources to merge.
              -> Int -- ^ The bound of the intermediate channel.
-             -> mi (Source mo a)
+             -> mo (ConduitT () a mi ())
 mergeSources sx bound = do
-     c <- liftSTM $ newTBMChan bound
-     refcount <- liftSTM . newTVar $ length sx
-     regs <- forM (map (transPipe lift) sx) $ \s -> Lifted.mask_ $ do
-       register . killThread <=< runResourceT $ resourceForkIO $ s $$ chanSink c writeTBMChan $ decRefcount refcount
-     return $ chanSource c readTBMChan
-                           (\chan -> do liftSTM $ closeTBMChan chan
-                                        mapM_ release regs)
+     return $ do 
+       (chkey, c) <- allocate (liftSTM $ newTBMChan bound)
+                              (liftSTM . closeTBMChan)
+       refcount <- liftSTM . newTVar $ length sx
+       st <- lift $ askUnliftIO
+       regs <- forM sx $ \s ->
+              register . killThread =<<
+                (liftIO $ forkIOWithUnmask $ \unmask ->
+                   (unmask $ unliftIO st $
+                      runConduit $ s .| chanSink c writeTBMChan)
+                    `Lifted.finally` (liftSTM $ decRefcount refcount c))
+       chanSource c readTBMChan
+       release chkey
+       traverse_ release regs
 
 -- | Combines two conduits with unbounded channels, creating a new conduit
 --   which pulls data from a mix of the two: whichever produces first.
 --
 --   The order of the new conduit's output is undefined, but it will be some
 --   combination of the two given conduits.
-(<=>) :: (MonadIO mi, MonadThrow mi, MonadIO mo, MonadBaseControl IO mi)
-      => Show i
-      => Conduit i (ResourceT mi) i
-      -> Conduit i (ResourceT mi) i
-      -> ResourceT mi (Conduit i mo i)
+(<=>) :: (MonadThrow mi, MonadIO mo, MonadUnliftIO mi)
+      => ConduitT i i (ResourceT mi) ()
+      -> ConduitT i i (ResourceT mi) ()
+      -> ResourceT mi (ConduitT i i mo ())
 sa <=> sb = mergeConduits [ sa, sb ] 16
 {-# INLINE (<=>) #-}
 
@@ -224,24 +235,37 @@
 -- @before 3.0
 --   Spawned threads are not guaranteed to be closed, This may happen if threads
 --   Conduit was closed before all threads have finished execution.
-mergeConduits :: (MonadIO mi, MonadIO mi, MonadThrow mi, MonadIO mo, MonadBaseControl IO mi)
-              => [Conduit i (ResourceT mi) o] -- ^ The conduits to merge.
+{-# DEPRECATED mergeConduits "This method will dissapear in the next version." #-}
+mergeConduits :: (MonadIO mo, MonadUnliftIO mi)
+              => [ConduitT i o (ResourceT mi) ()] -- ^ The conduits to merge.
               -> Int -- ^ The bound for the channels.
-              -> ResourceT mi (Conduit i mo o)
+              -> ResourceT mi (ConduitT i o mo ())
 mergeConduits conduits bound = do
         let len = length conduits
         refcount <- liftSTM $ newTVar len
-        iChannels <- replicateM len $ liftSTM $ newTBMChan bound
-        oChannel <- liftSTM $ newTBMChan bound
+        (iregs, iChannels) <- fmap unzip 
+             $ replicateM len $ allocate (liftSTM $ newTBMChan bound)
+                                          (liftSTM . closeTBMChan)
+        (oreg, oChannel) <- allocate (liftSTM $ newTBMChan bound)
+                                     (liftSTM . closeTBMChan)
         regs <- forM (zip iChannels conduits)
             $ \(iChannel, conduit) -> Lifted.mask_ $
               register . killThread <=< resourceForkIO
-                $  sourceTBMChan iChannel
-                $$ conduit
-                =$ chanSink oChannel writeTBMChan (decRefcount refcount)
+                $ (runConduit $ sourceTBMChan iChannel
+                             .| conduit
+                             .| chanSink oChannel writeTBMChan)
+                     `finally` 
+                        (liftIO $ atomically $ decRefcount refcount oChannel >> closeTBMChan iChannel)
+        treg <- register $ do 
+          traverse_ release regs
+          traverse_ release iregs
+          release oreg
         return
-            $  toConsumer (chanSink iChannels writeTBMChans (mapM_ closeTBMChan))
-            >> toProducer (chanSource oChannel readTBMChan (\c -> do atomically $ closeTBMChan c
-                                                                     mapM_ release regs))
+            $  do chanSink iChannels writeTBMChans
+                  -- release internal channels effiniently closing input channels
+                  traverse_ release iregs
+                  chanSource oChannel readTBMChan
+                  -- release internals everything
+                  release treg
   where
     writeTBMChans channels a = forM_ channels $ \c -> writeTBMChan c a
diff --git a/Data/Conduit/TQueue.hs b/Data/Conduit/TQueue.hs
--- a/Data/Conduit/TQueue.hs
+++ b/Data/Conduit/TQueue.hs
@@ -14,7 +14,7 @@
 --
 --
 --   Here is short description of data structures:
---     
+--
 --     * TQueue   - unbounded infinite queue
 --
 --     * TBQueue  - bounded infinite queue
@@ -24,7 +24,7 @@
 --     * TBMQueue - bounded finite (closable) queue
 --
 -- Caveats
--- 
+--
 --   Infinite operations means that source doesn't know when stream is
 --   ended so one need to use other methods of finishing stream like
 --   sending an exception or finish conduit in downstream.
@@ -62,35 +62,37 @@
 
 -- | A simple wrapper around a "TQueue". As data is pushed into the queue, the
 --   source will read it and pass it down the conduit pipeline.
-sourceTQueue :: MonadIO m => TQueue a -> Source m a
+sourceTQueue :: MonadIO m => TQueue a -> ConduitT z a m ()
 sourceTQueue q = forever $ liftSTM (readTQueue q) >>= yield
 
 -- | A simple wrapper around a "TQueue". As data is pushed into this sink, it
 --   will magically begin to appear in the queue.
-sinkTQueue :: MonadIO m => TQueue a -> Sink a m ()
+sinkTQueue :: MonadIO m => TQueue a -> ConduitT a z m ()
 sinkTQueue q = CL.mapM_ (liftSTM . writeTQueue q)
 
 -- | A simple wrapper around a "TBQueue". As data is pushed into the queue, the
 --   source will read it and pass it down the conduit pipeline.
-sourceTBQueue :: MonadIO m => TBQueue a -> Source m a
+sourceTBQueue :: MonadIO m => TBQueue a -> ConduitT z a m ()
 sourceTBQueue q = forever $ liftSTM (readTBQueue q) >>= yield
 
 -- | A simple wrapper around a "TBQueue". As data is pushed into this sink, it
---   will magically begin to appear in the queue. Boolean argument is used
---   to specify if queue should be closed when the sink is closed.
-sinkTBQueue :: MonadIO m => TBQueue a -> Sink a m ()
+--   will magically begin to appear in the queue.
+sinkTBQueue :: MonadIO m => TBQueue a -> ConduitT a z m ()
 sinkTBQueue q = CL.mapM_ (liftSTM . writeTBQueue q)
 
 -- | A convenience wrapper for creating a source and sink TBQueue of the given
 --   size at once, without exposing the underlying queue.
-entangledPair :: MonadIO m => Int -> m (Source m a, Sink a m ())
+--
+-- Returns release key that can be used for premature close of the communication
+-- channel, otherwise channel will be closed when the ResourceT scope will be closed.
+entangledPair :: MonadIO m => Int -> m (ConduitT z a m (), ConduitT a l m ())
 entangledPair size = liftM (liftM2 (,) sourceTBQueue sinkTBQueue) $
-    liftIO $ atomically $ newTBQueue size
+  liftIO $ atomically $ newTBQueue size
 
 -- | A simple wrapper around a "TMQueue". As data is pushed into the queue, the
 --   source will read it and pass it down the conduit pipeline. When the
 --   queue is closed, the source will close also.
-sourceTMQueue :: MonadIO m => TMQueue a -> Source m a
+sourceTMQueue :: MonadIO m => TMQueue a -> ConduitT z a m ()
 sourceTMQueue q =
     loop
   where
@@ -98,23 +100,19 @@
         mx <- liftSTM $ readTMQueue q
         case mx of
             Nothing -> return ()
-            Just x -> yieldOr x close >> loop
-    close = liftSTM $ closeTMQueue q
+            Just x -> yield x >> loop
 
 -- | A simple wrapper around a "TMQueue". As data is pushed into this sink, it
 --   will magically begin to appear in the queue.
 sinkTMQueue :: MonadIO m
             => TMQueue a
-            -> Bool -- ^ Should the queue be closed when the sink is closed?
-            -> Sink a m ()
-sinkTMQueue q shouldClose = do
-    CL.mapM_ (liftSTM . writeTMQueue q)
-    when shouldClose (liftSTM $ closeTMQueue q)
+            -> ConduitT a z m ()
+sinkTMQueue q = CL.mapM_ (liftSTM . writeTMQueue q)
 
 -- | A simple wrapper around a "TBMQueue". As data is pushed into the queue, the
 --   source will read it and pass it down the conduit pipeline. When the
 --   queue is closed, the source will close also.
-sourceTBMQueue :: MonadIO m => TBMQueue a -> Source m a
+sourceTBMQueue :: MonadIO m => TBMQueue a -> ConduitT z a m ()
 sourceTBMQueue q =
     loop
   where
@@ -122,18 +120,14 @@
         mx <- liftSTM $ readTBMQueue q
         case mx of
             Nothing -> return ()
-            Just x -> yieldOr x close >> loop
-    close = liftSTM $ closeTBMQueue q
+            Just x -> yield x >> loop
 
 -- | A simple wrapper around a "TBMQueue". As data is pushed into this sink, it
 --   will magically begin to appear in the queue.
 sinkTBMQueue :: MonadIO m
              => TBMQueue a
-             -> Bool -- ^ Should the queue be closed when the sink is closed?
-             -> Sink a m ()
-sinkTBMQueue q shouldClose = do
-    CL.mapM_ (liftSTM . writeTBMQueue q)
-    when shouldClose (liftSTM $ closeTBMQueue q)
+             -> ConduitT a z m ()
+sinkTBMQueue q = CL.mapM_ (liftSTM . writeTBMQueue q)
 
 
 liftSTM :: forall (m :: * -> *) a. MonadIO m => STM a -> m a
diff --git a/Data/Conduit/Utils.hs b/Data/Conduit/Utils.hs
--- a/Data/Conduit/Utils.hs
+++ b/Data/Conduit/Utils.hs
@@ -1,8 +1,8 @@
 {-# LANGUAGE FunctionalDependencies #-}
 {-# LANGUAGE KindSignatures #-}
 {-# LANGUAGE FlexibleInstances #-}
-{-| 
-This module provide different utility functions that allow to use safe higher 
+{-|
+This module provide different utility functions that allow to use safe higher
 level usage.
 
 Conduit pairs allow creation of an internal datastructure that acts as a bridge,
@@ -25,8 +25,8 @@
 >    mkSource = sourceTBMQueue
 >    mkSink   = flip sinkTBMQueue True
 
-  * Use "pair" or "pairBounded" to create a bridge. Because bridge data structure 
-    is hidden and not seen in parameters, we need proxy type to help compiler to 
+  * Use "pair" or "pairBounded" to create a bridge. Because bridge data structure
+    is hidden and not seen in parameters, we need proxy type to help compiler to
     choose type, we use "Proxy2" for that.
 
 >  pairTBMQueue = pairBounded (proxy2 :: Proxy2 TBMQueue a)
@@ -40,15 +40,15 @@
   conduit code.
 
 This package provides predefined pairs for all STM types that are used
-in the package. 
+in the package.
 
 -}
 module Data.Conduit.Utils
-  ( 
+  (
   -- * Conduit pairs
   -- ** Low level functions
     pairBounded   -- MonadIO m => m (Source m a, Sink m a ())
-  , pair          -- MonadIO m => Int -> m (Source m a, Sink m a ()) 
+  , pair          -- MonadIO m => Int -> m (Source m a, Sink m a ())
   -- ** Classes
   , UnboundedStream(..)
   , BoundedStream(..)
@@ -90,14 +90,14 @@
 
 -- | Class for structures that can handle unbounded stream of values.
 -- Such streams break conduit assumptions that constant memory will be
--- used, because if receiver is slower then sender than values will 
+-- used, because if receiver is slower then sender than values will
 -- be accumulated.
 class UnboundedStream i o | i -> o where
   mkUStream :: i a -> IO (o a)
 
 -- | Class for structures that can handle bounded stream of values i.e.
--- there is exists 'Int' value that sets an upper limit on the number 
--- of values that can be handled by structure. Exact meaning of this 
+-- there is exists 'Int' value that sets an upper limit on the number
+-- of values that can be handled by structure. Exact meaning of this
 -- limit may depend on the carrier type.
 class BoundedStream i o | i -> o where
   mkBStream :: i a -> Int -> IO (o a)
@@ -105,14 +105,14 @@
 -- | Class that describes how we can make conduit out of the carrier
 -- value.
 class MonadIO m => IsConduit m (x :: * -> *) where
-  mkSink   :: x a -> Sink a m ()
-  mkSource :: x a -> Source m a
+  mkSink   :: x a -> ConduitT a Void m ()
+  mkSource :: x a -> ConduitT () a m ()
 
 -- | Create bounded conduit pair, see "BoundedStream" class description.
 pairBounded :: (MonadIO m, IsConduit m o, BoundedStream i o)
             => i a    -- ^ Type description.
             -> Int    -- ^ Conduit size.
-            -> m (Source m a, Sink a m ())
+            -> m (ConduitT () a m (), ConduitT a Void m ())
 pairBounded p s = do
   q <- liftIO $ mkBStream p s
   return (mkSource q, mkSink q)
@@ -120,7 +120,7 @@
 -- | Create unbounded pair, see "UnboundedStream" class description.
 pair :: (MonadIO m, IsConduit m o, UnboundedStream i o)
      => i a   -- ^ Type description.
-     -> m (Source m a, Sink a m ())
+     -> m (ConduitT () a m (), ConduitT a Void m ())
 pair p = do
   q <- liftIO $ mkUStream p
   return (mkSource q, mkSink q)
@@ -152,11 +152,11 @@
 
 instance MonadIO m => IsConduit m TBMQueue where
   mkSource = sourceTBMQueue
-  mkSink   = flip sinkTBMQueue True
+  mkSink   = sinkTBMQueue
 
 instance MonadIO m => IsConduit m TMQueue where
   mkSource = sourceTMQueue
-  mkSink   = flip sinkTMQueue True
+  mkSink   = sinkTMQueue
 
 instance MonadIO m => IsConduit m TQueue where
   mkSource = sourceTQueue
@@ -164,11 +164,11 @@
 
 instance MonadIO m => IsConduit m TBMChan where
   mkSource = sourceTBMChan
-  mkSink   = flip sinkTBMChan True
+  mkSink   = sinkTBMChan
 
 instance MonadIO m => IsConduit m TMChan where
   mkSource = sourceTMChan
-  mkSink   = flip sinkTMChan True
+  mkSink   = sinkTMChan
 
 -------------------------------------------------------------------------------
 -- Specialized functions
@@ -179,12 +179,12 @@
 -- is not closable then there is no way to notify receiver side that bridge
 -- is closed, so it's possible to use it only in infinite streams of when
 -- some other mechanism of notification is used.
-pairTQueue, pairTMQueue, pairTMChan :: MonadIO m => m (Source m a, Sink a m ())
+pairTQueue, pairTMQueue, pairTMChan :: MonadIO m => m (ConduitT () a m (), ConduitT a Void m ())
 pairTQueue   = pair (proxy2 :: Proxy2 TQueue a)
 pairTMQueue  = pair (proxy2 :: Proxy2 TMQueue a)
 pairTMChan   = pair (proxy2 :: Proxy2 TMChan a)
 
-pairTBQueue, pairTBMQueue, pairTBMChan :: MonadIO m => Int -> m (Source m a, Sink a m ())
+pairTBQueue, pairTBMQueue, pairTBMChan :: MonadIO m => Int -> m (ConduitT () a m (), ConduitT a Void m ())
 pairTBQueue  = pairBounded (proxy2 :: Proxy2 TBQueue a)
 pairTBMQueue = pairBounded (proxy2 :: Proxy2 TBMQueue a)
 pairTBMChan  = pairBounded (proxy2 :: Proxy2 TBMChan a)
diff --git a/stm-conduit.cabal b/stm-conduit.cabal
--- a/stm-conduit.cabal
+++ b/stm-conduit.cabal
@@ -1,7 +1,8 @@
 Name:                stm-conduit
-Version:             3.0.0
+Version:             4.0.0
 Synopsis:            Introduces conduits to channels, and promotes using conduits concurrently.
 Description:         Provides two simple conduit wrappers around STM channels - a source and a sink.
+                     
 Homepage:            https://github.com/cgaebel/stm-conduit
 License:             BSD3
 License-file:        LICENSE
@@ -11,8 +12,7 @@
 
 Build-type:          Simple
 
-tested-with: GHC >= 7.6 && < 7.11
--- selects GHC 7.4.1, 7.4.2, 7.6.1, 7.6.2, and 7.6.3
+tested-with: GHC == 8.0.1, GHC == 8.2.1
 
 Cabal-version:       >=1.8
 
@@ -32,19 +32,15 @@
       , stm                 == 2.4.*
       , stm-chans           >= 2.0 && < 3.1
       , cereal              >= 0.4.0.1
-      , cereal-conduit      >= 0.7.2
-      , conduit             >= 1.0 && < 1.3
-      , conduit-combinators >= 0.3
-      , conduit-extra       >= 1.0 && < 1.2
+      , cereal-conduit      >= 0.8
+      , conduit             >= 1.0 && < 1.4
+      , conduit-extra       >= 1.0 && < 1.4
       , directory           >= 1.1
-      , resourcet           >= 0.3 && < 1.2
+      , exceptions
+      , resourcet           >= 0.3 && < 1.3
       , async               >= 2.0.1
-      , monad-control       >= 0.3.2
       , monad-loops         >= 0.4.2
-      , lifted-base         >= 0.2.1
-      , lifted-async        >= 0.1
-      , void                >= 0.7
-      , ghc-prim
+      , unliftio            >= 0.2.0 && < 0.3.0
 
     ghc-options: -Wall -fwarn-tabs -fwarn-unused-imports
 
@@ -74,11 +70,11 @@
       , stm
       , stm-conduit
       , conduit
-      , conduit-combinators >= 0.3
       , transformers
       , stm-chans
       , resourcet
       , directory
+      , unliftio
 
 source-repository head
     type:     git
diff --git a/test/DocTests.hs b/test/DocTests.hs
--- a/test/DocTests.hs
+++ b/test/DocTests.hs
@@ -4,11 +4,11 @@
 
 main :: IO ()
 main = doctest [
-  "-packageghc"
-  , "-isrc"
-  , "-idist/build/autogen/"
-  , "-optP-include"
-  , "-optPdist/build/autogen/cabal_macros.h"
-  , "-cpp"
-  , "Data/Conduit/Async/Composition.hs"
+--  "-packageghc"
+--  , "-isrc"
+--  , "-idist/build/autogen/"
+--  , "-optP-include"
+--  , "-optPdist/build/autogen/cabal_macros.h"
+--  , "-cpp"
+--  , "Data/Conduit/Async/Composition.hs"
   ]
diff --git a/test/Test.hs b/test/Test.hs
--- a/test/Test.hs
+++ b/test/Test.hs
@@ -1,4 +1,4 @@
-{-# LANGUAGE ScopedTypeVariables, RankNTypes, FlexibleContexts #-}
+{-# LANGUAGE ScopedTypeVariables, RankNTypes, FlexibleContexts, BangPatterns #-}
 
 module Main ( main ) where
 
@@ -12,7 +12,7 @@
 import Test.HUnit
 
 import qualified Control.Monad as Monad
-import Control.Monad.Trans.Resource (runResourceT)
+import Control.Monad.Trans.Resource (runResourceT, allocate, release)
 import Control.Concurrent (forkIO, threadDelay)
 import Control.Concurrent.STM
 import Control.Concurrent.STM.TMQueue
@@ -23,12 +23,13 @@
 import Data.Conduit.TQueue
 import System.Directory
 import Conduit
+import qualified UnliftIO as Lifted
 
 main = defaultMain tests
 
 tests = [
         testGroup "Behaves to spec" [
-                  testCase "simpleList using TMChan" test_simpleList
+                testCase "simpleList using TMChan" test_simpleList
                 , testCase "simpleList using TQueue" test_simpleQueue
                 , testCase "simpleList using TMQueue" test_simpleMQueue
             ],
@@ -48,39 +49,65 @@
             ]
     ]
 
-test_simpleList = do chan <- atomically $ newTMChan
-                     forkIO . runResourceT $ sourceList testList $$ sinkTMChan chan True
-                     lst' <- runResourceT $ sourceTMChan chan $$ consume
-                     assertEqual "for the numbers [1..10000]," testList lst'
-                     closed <- atomically $ isClosedTMChan chan
-                     assertBool "channel is closed after running" closed
+test_simpleList :: IO ()
+test_simpleList =
+   -- We start global resource region because we want to protect
+   -- the channel. This is not strictly required under some cases
+   -- but is needed in order to have exception safety. We need
+   -- to cleanup properly in case of exception, but it depending
+   -- on the use case it may be implemented by the other means.
+   runResourceT $ do
+     (reg, chan) <- allocate (newTMChanIO)
+                             (atomically . closeTMChan)
+     _ <- Lifted.async $ do
+       runConduit $ sourceList testList .| sinkTMChan chan
+       -- We have to release resource explicitly in order to
+       -- show that no data will be sent here ever.
+       release reg
+      
+     lst' <- runConduit $ sourceTMChan chan .| consume
+     liftIO $ do 
+       assertEqual "for the numbers [1..10000]," testList lst'
+       closed <- atomically $ isClosedTMChan chan
+       assertBool "channel is closed after running" closed
     where
         testList = [1..10000]
 
-test_simpleQueue = do q <- atomically $ newTQueue
-                      forkIO . runResourceT $ sourceList testList $$ sinkTQueue q
-                      lst'  <- runResourceT $ sourceTQueue q $$ CL.take (length testList)
-                      assertEqual "for the numbers [1..10000]," testList lst'
+test_simpleQueue :: IO ()
+test_simpleQueue = runResourceT $ do
+  q <- liftIO $ atomically $ newTQueue
+  liftIO $ forkIO . runConduit $ sourceList testList .| sinkTQueue q
+  lst'  <- runConduit $ sourceTQueue q .| CL.take (length testList)
+  liftIO $ assertEqual "for the numbers [1..10000]," testList lst'
     where
         testList = [1..10000]
 
-test_simpleMQueue = do q <- atomically $ newTMQueue
-                       forkIO . runResourceT $ sourceList testList $$ sinkTMQueue q True
-                       lst' <- runResourceT $ sourceTMQueue q $$ consume
-                       assertEqual "for the numbers [1..10000]," testList lst'
-                       closed <- atomically $ isClosedTMQueue q
-                       assertBool "channel is closed after running" closed
-    where
-        testList = [1..10000]
+test_simpleMQueue :: IO ()
+test_simpleMQueue = runResourceT $ do
+  (reg, q) <- allocate (newTMQueueIO) (atomically . closeTMQueue)
+  _ <- Lifted.async $ do
+        runConduit $ sourceList testList .| sinkTMQueue q
+        release reg
+        
+  lst' <- runConduit $ sourceTMQueue q .| consume
+  liftIO $ do
+    assertEqual "for the numbers [1..10000]," testList lst'
+    closed <- atomically $ isClosedTMQueue q
+    assertBool "channel is closed after running" closed
+   where
+     testList = [1..10000]
 
-test_multipleWriters = do xs <- runResourceT $ do
-                            ms <- mergeSources [ sourceList ([1..10]::[Integer])
-                                               , sourceList ([11..20]::[Integer])
-                                               ] 3
-                            ms $$ consume
-                          assertEqual "for the numbers [1..10] and [11..20]," [1..20] $ sort xs
+test_multipleWriters :: IO ()
+test_multipleWriters = do
+  xs <- runResourceT $ do
+          ms <- mergeSources [ sourceList ([1..10]::[Integer])
+                             , sourceList ([11..20]::[Integer])
+                             ] 3
+          runConduit $ ms .| consume
+  assertEqual "for the numbers [1..10] and [11..20]," [1..20] $ sort xs
 
-test_asyncOperator = do sum'  <- CL.sourceList [1..n] $$ CL.fold (+) 0
+test_asyncOperator :: IO ()
+test_asyncOperator = do sum'  <- runConduit $ 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''
@@ -89,10 +116,12 @@
         n = 100
         sum = (n * (n+1)) `div` 2
 
+test_buffer :: IO ()
 test_buffer = do
     sum' <- buffer 128 (CL.sourceList [1..100]) (CL.fold (+) 0)
     assertEqual "sum computed using buffer" sum' (5050 :: Integer)
 
+test_multi_buffer :: IO ()
 test_multi_buffer = do
     sumDoubles <- CL.sourceList [1..100] $$& mapC (* 2) $=& CL.fold (+) 0
     assertEqual "sum of doubles computed using two buffers" sumDoubles (10100 :: Integer)
@@ -100,36 +129,43 @@
 -- When we're testing file-buffering, we have to make sure to consume
 -- slowly enough to ensure the incoming data piles up enough to be
 -- flushed to disk..
-slowDown :: (MonadIO m) => Int -> Conduit x m x
+slowDown :: (MonadIO m) => Int -> ConduitT x x m ()
 slowDown delay = awaitForever $ \x -> do
   liftIO $ threadDelay delay
   yield x
 
+aLot, aLittle :: Int
 aLot = 10000
 aLittle = 5000
 
+test_bufferToFile :: IO ()
 test_bufferToFile = do
     tempDir <- getTemporaryDirectory
-    sum' <- runResourceT $ bufferToFile 16 (Just 25) tempDir (CL.sourceList [1 :: Int .. 100]) (slowDown aLittle $= CL.fold (+) 0)
+    sum' <- runResourceT $ bufferToFile 16 (Just 25) tempDir (CL.sourceList [1 :: Int .. 100]) (slowDown aLittle .| CL.fold (+) 0)
     assertEqual "sum computed using bufferToFile" sum' 5050
 
+test_multi_bufferToFile :: IO ()
 test_multi_bufferToFile = do
     tempDir <- getTemporaryDirectory
     sumDoubles <- let buf c = bufferToFile' 16 (Just 25) tempDir c -- "c" avoids monomorphism restriction
-                  in runResourceT $ runCConduit $ CL.sourceList [1 :: Int .. 100] `buf` (slowDown aLittle $= mapC (* 2)) `buf` (slowDown aLot $= CL.fold (+) 0)
+                  in runResourceT $ runCConduit $ CL.sourceList [1 :: Int .. 100] `buf` (slowDown aLittle .| mapC (* 2)) `buf` (slowDown aLot .| CL.fold (+) 0)
     assertEqual "sum of doubles computed using bufferToFile" sumDoubles 10100
 
+test_mixed_buffer :: IO ()
 test_mixed_buffer = do
     tempDir <- getTemporaryDirectory
-    sumDoubles <- let buf = bufferToFile' 16 (Just 25) tempDir
-                  in runResourceT $ CL.sourceList [1 :: Int .. 100] $$& mapC (* 2) `buf` (slowDown aLittle $= CL.fold (+) 0)
+    sumDoubles <-
+      let buf = bufferToFile' 16 (Just 25) tempDir
+      in runResourceT $
+           CL.sourceList [1 :: Int .. 100] $$& mapC (* 2) `buf` (slowDown aLittle .| CL.fold (+) 0)
     assertEqual "sum of doubles computed using mixed buffers" sumDoubles 10100
     sumTriples <- let buf = bufferToFile' 16 (Just 25) tempDir
-                  in runResourceT $ CL.sourceList [1 :: Int .. 100] `buf` (slowDown aLittle $= mapC (* 3)) $$& CL.fold (+) 0
+                  in runResourceT $ CL.sourceList [1 :: Int .. 100] `buf` (slowDown aLittle .| mapC (* 3)) $$& CL.fold (+) 0
     assertEqual "sum of triples computed using mixed buffers" sumTriples 15150
 
+test_gatherFrom :: IO ()
 test_gatherFrom = do
-    sum' <- gatherFrom 128 gen $$ CL.fold (+) 0
+    sum' <- runConduit $ gatherFrom 128 gen .| CL.fold (+) 0
     assertEqual "sum computed using gatherFrom" sum' 5050
   where
     gen queue = Monad.void $ Monad.foldM f queue [1..100]
@@ -138,16 +174,18 @@
             atomically $ writeTBQueue q x
             return q
 
+test_drainTo :: IO ()
 test_drainTo = do
-    sum' <- CL.sourceList [1..100] $$ drainTo 128 (go 0)
+    sum' <- runConduit $ CL.sourceList [1..100] .| drainTo 128 (go 0)
     assertEqual "sum computed using drainTo" sum' 5050
   where
-    go acc queue = do
+    go !acc queue = do
         mres <- atomically $ readTBQueue queue
         case mres of
             Nothing  -> return acc
             Just res -> go (acc + res) queue
 
+test_mergeConduits :: IO ()
 test_mergeConduits = do
     merged <- runResourceT $ mergeConduits
                 [ CL.map (* 2)
@@ -156,5 +194,5 @@
     let
       input = [1..10]
       expected = Prelude.map (2 *) input ++ tail (Prelude.scanl (+) 0 input)
-    xs <- sourceList ([1..10] :: [Integer]) $$ merged =$ consume
+    xs <- runConduit $ sourceList ([1..10] :: [Integer]) .|  merged .| consume
     assertEqual "merged results" (sort expected) (sort xs)