diff --git a/Data/Conduit/Classy.hs b/Data/Conduit/Classy.hs
new file mode 100644
--- /dev/null
+++ b/Data/Conduit/Classy.hs
@@ -0,0 +1,323 @@
+{-# LANGUAGE CPP #-}
+{-# LANGUAGE TypeFamilies #-}
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE NoImplicitPrelude #-}
+{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE FunctionalDependencies #-}
+{-# LANGUAGE UndecidableInstances #-}
+-- | Note: This module is experimental, and might be modified at any time.
+-- Caveat emptor!
+module Data.Conduit.Classy
+    ( module Data.Conduit.Classy
+    , C.ResumableSource
+    , C.runResourceT
+    , C.Flush (..)
+    , C.ResourceT
+    , C.unwrapResumable
+    ) where
+
+import Prelude (Monad (..), Functor (..), ($), const, IO, Maybe, Either, Bool, (.), either)
+import Data.Void (Void)
+import Control.Applicative (Applicative (..))
+import qualified Data.Conduit as C
+import Data.Conduit.Internal (Pipe (PipeM))
+import Control.Monad.Trans.Class (MonadTrans (..))
+import Control.Monad.Trans.Resource (allocate, release, MonadThrow, MonadResource, ResourceT)
+import Control.Monad.Trans.Control (liftWith, restoreT, MonadTransControl)
+import Control.Monad.IO.Class (MonadIO)
+import Data.Monoid (Monoid (..))
+
+import Control.Monad.Trans.Identity ( IdentityT)
+import Control.Monad.Trans.List     ( ListT    )
+import Control.Monad.Trans.Maybe    ( MaybeT   )
+import Control.Monad.Trans.Error    ( ErrorT, Error)
+import Control.Monad.Trans.Reader   ( ReaderT  )
+import Control.Monad.Trans.State    ( StateT   )
+import Control.Monad.Trans.Writer   ( WriterT  )
+import Control.Monad.Trans.RWS      ( RWST     )
+
+import qualified Control.Monad.Trans.RWS.Strict    as Strict ( RWST   )
+import qualified Control.Monad.Trans.State.Strict  as Strict ( StateT )
+import qualified Control.Monad.Trans.Writer.Strict as Strict ( WriterT )
+
+-- | Provides a stream of output values, without consuming any input or
+-- producing a final result.
+--
+-- Since 0.6.0
+type Source m o = SourceM o m ()
+
+newtype SourceM o m r = SourceM { unSourceM :: Pipe () () o () m r }
+    deriving (Functor, Applicative, Monad, MonadTrans, MonadIO, ResourcePipe, MonadThrow)
+
+instance Monad m => Monoid (SourceM o m ()) where
+    mempty = return ()
+    mappend = (>>)
+
+-- | Consumes a stream of input values and produces a stream of output values,
+-- without producing a final result.
+--
+-- Since 0.6.0
+type Conduit i m o = ConduitM i o m ()
+
+newtype ConduitM i o m r = ConduitM { unConduitM :: Pipe i i o () m r }
+    deriving (Functor, Applicative, Monad, MonadTrans, MonadIO, ResourcePipe, MonadThrow)
+
+instance Monad m => Monoid (ConduitM i o m ()) where
+    mempty = return ()
+    mappend = (>>)
+
+-- | Consumes a stream of input values and produces a final result, without
+-- producing any output.
+--
+-- Since 0.6.0
+newtype Sink i m r = Sink { unSink :: Pipe i i Void () m r }
+    deriving (Functor, Applicative, Monad, MonadTrans, MonadIO, ResourcePipe, MonadThrow)
+
+instance Monad m => Monoid (Sink i m ()) where
+    mempty = return ()
+    mappend = (>>)
+
+class (Monad m, Monad (PipeMonad m)) => IsPipe m where
+    type PipeInput m
+    type PipeTerm m
+    type PipeOutput m
+    type PipeMonad m :: * -> *
+
+    -- | Wait for a single input value from upstream, terminating immediately if no
+    -- data is available.
+    --
+    -- Since 0.5.0
+    await :: m (Maybe (PipeInput m))
+
+    -- | This is similar to @await@, but will return the upstream result value as
+    -- @Left@ if available.
+    --
+    -- Since 0.5.0
+    awaitE :: m (Either (PipeTerm m) (PipeInput m))
+
+    -- | Provide a single piece of leftover input to be consumed by the next pipe
+    -- in the current monadic binding.
+    --
+    -- /Note/: it is highly encouraged to only return leftover values from input
+    -- already consumed from upstream.
+    --
+    -- Since 0.5.0
+    leftover :: PipeInput m -> m ()
+
+    -- | Send a single output value downstream. If the downstream @Pipe@
+    -- terminates, this @Pipe@ will terminate as well.
+    --
+    -- Since 0.5.0
+    yield :: PipeOutput m -> m ()
+
+    -- | Similar to @yield@, but additionally takes a finalizer to be run if the
+    -- downstream @Pipe@ terminates.
+    --
+    -- Since 0.5.0
+    yieldOr :: PipeOutput m -> PipeMonad m () -> m ()
+
+    liftPipeMonad :: PipeMonad m a -> m a
+
+    -- | Add some code to be run when the given @Pipe@ cleans up.
+    --
+    -- Since 0.4.1
+    addCleanup :: (Bool -> PipeMonad m ()) -- ^ @True@ if @Pipe@ ran to completion, @False@ for early termination.
+               -> m r
+               -> m r
+
+instance (Monad m, l ~ i) => IsPipe (Pipe l i o u m) where
+    type PipeInput (Pipe l i o u m) = i
+    type PipeTerm (Pipe l i o u m) = u
+    type PipeOutput (Pipe l i o u m) = o
+    type PipeMonad (Pipe l i o u m) = m
+
+    await = C.await
+    {-# INLINE [1] await #-}
+
+    awaitE = C.awaitE
+    {-# INLINE [1] awaitE #-}
+
+    leftover = C.leftover
+    {-# INLINE [1] leftover #-}
+
+    yield = C.yield
+    {-# INLINE yield #-}
+
+    yieldOr = C.yieldOr
+    {-# INLINE yieldOr #-}
+
+    liftPipeMonad = lift
+
+    addCleanup = C.addCleanup
+
+instance Monad m => IsPipe (SourceM o m) where
+    type PipeInput (SourceM o m) = ()
+    type PipeTerm (SourceM o m) = ()
+    type PipeOutput (SourceM o m) = o
+    type PipeMonad (SourceM o m) = m
+
+    await = SourceM await
+    {-# INLINE await #-}
+
+    awaitE = SourceM awaitE
+    {-# INLINE awaitE #-}
+
+    leftover = SourceM . leftover
+    {-# INLINE leftover #-}
+
+    yield = SourceM . yield
+    {-# INLINE yield #-}
+
+    yieldOr a = SourceM . yieldOr a
+    {-# INLINE yieldOr #-}
+
+    liftPipeMonad = lift
+    {-# INLINE liftPipeMonad #-}
+
+    addCleanup c (SourceM p) = SourceM (addCleanup c p)
+    {-# INLINE addCleanup #-}
+
+instance Monad m => IsPipe (ConduitM i o m) where
+    type PipeInput (ConduitM i o m) = i
+    type PipeTerm (ConduitM i o m) = ()
+    type PipeOutput (ConduitM i o m) = o
+    type PipeMonad (ConduitM i o m) = m
+
+    await = ConduitM await
+    {-# INLINE await #-}
+
+    awaitE = ConduitM awaitE
+    {-# INLINE awaitE #-}
+
+    leftover = ConduitM . leftover
+    {-# INLINE leftover #-}
+
+    yield = ConduitM . yield
+    {-# INLINE yield #-}
+
+    yieldOr a = ConduitM . yieldOr a
+    {-# INLINE yieldOr #-}
+
+    liftPipeMonad = lift
+    {-# INLINE liftPipeMonad #-}
+
+    addCleanup c (ConduitM p) = ConduitM (addCleanup c p)
+    {-# INLINE addCleanup #-}
+
+instance Monad m => IsPipe (Sink i m) where
+    type PipeInput (Sink i m) = i
+    type PipeTerm (Sink i m) = ()
+    type PipeOutput (Sink i m) = Void
+    type PipeMonad (Sink i m) = m
+
+    await = Sink await
+    {-# INLINE await #-}
+
+    awaitE = Sink awaitE
+    {-# INLINE awaitE #-}
+
+    leftover = Sink . leftover
+    {-# INLINE leftover #-}
+
+    yield = Sink . yield
+    {-# INLINE yield #-}
+
+    yieldOr a = Sink . yieldOr a
+    {-# INLINE yieldOr #-}
+
+    liftPipeMonad = lift
+    {-# INLINE liftPipeMonad #-}
+
+    addCleanup c (Sink p) = Sink (addCleanup c p)
+    {-# INLINE addCleanup #-}
+
+class (IsPipe m, MonadResource (PipeMonad m), MonadIO m) => ResourcePipe m where
+    -- | Perform some allocation and run an inner @Pipe@. Two guarantees are given
+    -- about resource finalization:
+    --
+    -- 1. It will be /prompt/. The finalization will be run as early as possible.
+    --
+    -- 2. It is exception safe. Due to usage of @resourcet@, the finalization will
+    --    be run in the event of any exceptions.
+    --
+    -- Since 0.5.0
+    bracketP :: IO a -> (a -> IO ()) -> (a -> m r) -> m r
+
+instance (l ~ i, MonadResource m) => ResourcePipe (Pipe l i o u m) where
+    bracketP alloc free inside = PipeM $ do
+        (key, seed) <- allocate alloc free
+        return $ addCleanup (const $ release key) (inside seed)
+
+#define GOALL(C, C2, T) instance C => IsPipe (T) where { type PipeInput (T) = PipeInput m; type PipeMonad (T) = PipeMonad m; type PipeTerm (T) = PipeTerm m; type PipeOutput (T) = PipeOutput m; await = lift await; awaitE = lift awaitE; leftover = lift . leftover; yield = lift . yield; yieldOr a = lift . yieldOr a; liftPipeMonad = lift . liftPipeMonad; addCleanup c r = liftWith (\run -> run $ addCleanup c r) >>= restoreT . return}; instance C2 => ResourcePipe (T) where { bracketP = controlBracketP }
+#define GO(T) GOALL(IsPipe m, ResourcePipe m, T m)
+#define GOX(X, T) GOALL((IsPipe m, X), (ResourcePipe m, X), T m)
+GO(IdentityT)
+GO(ListT)
+GO(MaybeT)
+GOX(Error e, ErrorT e)
+GO(ReaderT r)
+GO(StateT s)
+GOX(Monoid w, WriterT w)
+GOX(Monoid w, RWST r w s)
+GOX(Monoid w, Strict.RWST r w s)
+GO(Strict.StateT s)
+GOX(Monoid w, Strict.WriterT w)
+GO(ResourceT)
+#undef GO
+#undef GOX
+#undef GOALL
+
+controlBracketP :: (ResourcePipe m, Monad (t m), MonadTransControl t)
+                => IO a -> (a -> IO ()) -> (a -> t m r) -> t m r
+controlBracketP alloc free inside = liftWith (\run -> bracketP alloc free (run . inside)) >>= restoreT . return
+
+-- | Wait for input forever, calling the given inner @Pipe@ for each piece of
+-- new input. Returns the upstream result type.
+--
+-- Since 0.5.0
+awaitForever :: IsPipe m
+             => (PipeInput m -> m r')
+             -> m (PipeTerm m)
+awaitForever inner =
+    self
+  where
+    self = awaitE >>= either return (\i -> inner i >> self)
+{-# INLINE [1] awaitForever #-}
+
+infixr 0 $$
+infixl 1 $=
+infixr 2 =$
+infixr 2 =$=
+infixr 0 $$+
+infixr 0 $$++
+infixr 0 $$+-
+
+($$) :: Monad m => Source m a -> Sink a m b -> m b
+SourceM src $$ Sink sink = src C.$$ sink
+{-# INLINE ($$) #-}
+
+($=) :: Monad m => Source m a -> Conduit a m b -> Source m b
+SourceM src $= ConduitM con = SourceM $ src C.$= con
+{-# INLINE ($=) #-}
+
+(=$=) :: Monad m => Conduit a m b -> Conduit b m c -> Conduit a m c
+ConduitM l =$= ConduitM r = ConduitM $ l C.=$= r
+{-# INLINE (=$=) #-}
+
+(=$) :: Monad m => Conduit a m b -> Sink b m c -> Sink a m c
+ConduitM l =$ Sink r = Sink $ l C.=$ r
+{-# INLINE (=$) #-}
+
+($$+) :: Monad m => Source m a -> Sink a m b -> m (C.ResumableSource m a, b)
+SourceM src $$+ Sink sink = src C.$$+ sink
+{-# INLINE ($$+) #-}
+
+($$++) :: Monad m => C.ResumableSource m a -> Sink a m b -> m (C.ResumableSource m a, b)
+rsrc $$++ Sink sink = rsrc C.$$++ sink
+{-# INLINE ($$++) #-}
+
+($$+-) :: Monad m => C.ResumableSource m a -> Sink a m b -> m b
+rsrc $$+- Sink sink = rsrc C.$$+- sink
+{-# INLINE ($$+-) #-}
diff --git a/Data/Conduit/Container.hs b/Data/Conduit/Container.hs
new file mode 100644
--- /dev/null
+++ b/Data/Conduit/Container.hs
@@ -0,0 +1,141 @@
+{-# LANGUAGE NoImplicitPrelude #-}
+{-# LANGUAGE TypeFamilies #-}
+{-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE FunctionalDependencies #-}
+-- | Note: This module is experimental, and might be modified at any time.
+-- Caveat emptor!
+module Data.Conduit.Container where
+
+import Prelude ((.), Maybe (..), Monad (..), fmap, maybe, seq, Either (..), const, either, (-), ($), Int, compare, Ordering (..), id)
+import qualified Prelude
+import Data.Conduit.Classy
+import qualified Data.ByteString as S
+import qualified Data.ByteString.Lazy as L
+import Data.Word (Word8)
+import Control.Monad (liftM)
+
+class Container c where
+    type Single c
+    type Multi c
+
+    toSource :: (IsPipe m, PipeOutput m ~ c) => Multi c -> m ()
+
+    headE :: (IsPipe m, PipeInput m ~ c) => m (Either (PipeTerm m) (Single c))
+    head :: (IsPipe m, PipeInput m ~ c) => m (Maybe (Single c))
+    head = liftM (either (const Nothing) Just) headE
+
+    fold :: (IsPipe m, PipeInput m ~ c) => (accum -> Single c -> accum) -> accum -> m accum
+    fold f =
+        loop
+      where
+        loop accum =
+            head >>= maybe (return accum) go
+          where
+            go a =
+                let accum' = f accum a
+                 in accum' `seq` loop accum'
+
+    foldM :: (IsPipe m, PipeInput m ~ c) => (accum -> Single c -> m accum) -> accum -> m accum
+    foldM f =
+        loop
+      where
+        loop accum =
+            head >>= maybe (return accum) go
+          where
+            go a = do
+                accum' <- f accum a
+                accum' `seq` loop accum'
+
+    mapM_ :: (IsPipe m, PipeInput m ~ c) => (Single c -> m ()) -> m (PipeTerm m)
+    mapM_ f =
+        loop
+      where
+        loop = headE >>= either return (\s -> f s >> loop)
+
+    drop :: (IsPipe m, PipeInput m ~ c) => Int -> m ()
+    drop 0 = return ()
+    drop i = head >>= maybe (return ()) (const $ drop (i - 1))
+
+    singleton :: Single c -> c
+    isolate :: (IsPipe m, PipeInput m ~ c, PipeOutput m ~ c) => Int -> m ()
+    isolate 0 = return ()
+    isolate i = head >>= maybe (return ()) (\x -> yield (singleton x) >> isolate (i - 1))
+    consume :: (IsPipe m, PipeInput m ~ c) => m (Multi c)
+    take :: (IsPipe m, PipeInput m ~ c) => Int -> m (Multi c)
+
+instance Container S.ByteString where
+    type Single S.ByteString = Word8
+    type Multi S.ByteString = L.ByteString
+
+    toSource = Prelude.mapM_ yield . L.toChunks
+
+    headE = do
+        ebs <- awaitE
+        case ebs of
+            Left t -> return (Left t)
+            Right bs ->
+                case S.uncons bs of
+                    Nothing -> headE
+                    Just (w, bs') -> leftover bs' >> return (Right w)
+
+    fold f =
+        loop
+      where
+        loop accum =
+            await >>= maybe (return accum) go
+          where
+            go bs =
+                let accum' = S.foldl' f accum bs
+                 in accum' `seq` loop accum'
+
+    mapM_ f =
+        loop
+      where
+        loop = awaitE >>= either return (\bs -> Prelude.mapM_ f (S.unpack bs) >> loop)
+
+    drop 0 = return ()
+    drop i = await >>= maybe (return ()) (\bs ->
+        case i `compare` S.length bs of
+            LT -> leftover $ S.drop i bs
+            EQ -> return ()
+            GT -> drop (i - S.length bs))
+
+    singleton = S.singleton
+    consume =
+        loop id
+      where
+        loop front = await >>= maybe (return $ L.fromChunks $ front []) (\bs -> loop $ front . (bs:))
+
+    take =
+        loop id
+      where
+        loop front 0 = return $ L.fromChunks $ front []
+        loop front i = await >>= maybe (return $ L.fromChunks $ front []) (\bs ->
+            case i `compare` S.length bs of
+                LT -> do
+                    let (x, y) = S.splitAt i bs
+                    leftover y
+                    return $ L.fromChunks $ front [x]
+                EQ -> return $ L.fromChunks $ front [bs]
+                GT -> loop (front . (bs:)) (i - S.length bs))
+
+newtype Singleton a = Singleton { unSingleton :: a }
+
+instance Container (Singleton a) where
+    type Single (Singleton a) = a
+    type Multi (Singleton a) = [a]
+
+    toSource = Prelude.mapM_ (yield . Singleton)
+
+    headE = liftM (fmap unSingleton) awaitE
+
+    singleton = Singleton
+    consume =
+        loop id
+      where
+        loop front = head >>= maybe (return (front [])) (\x -> loop (front . (x:)))
+    take =
+        loop id
+      where
+        loop front 0 = return (front [])
+        loop front i = head >>= maybe (return (front [])) (\x -> loop (front . (x:)) (i - 1))
diff --git a/classy-prelude-conduit.cabal b/classy-prelude-conduit.cabal
--- a/classy-prelude-conduit.cabal
+++ b/classy-prelude-conduit.cabal
@@ -1,5 +1,5 @@
 name:                classy-prelude-conduit
-version:             0.4.1
+version:             0.4.2
 synopsis:            conduit instances for classy-prelude
 description:         conduit instances for classy-prelude
 homepage:            https://github.com/snoyberg/classy-prelude
@@ -13,11 +13,32 @@
 
 library
   exposed-modules:     ClassyPrelude.Conduit
+                       Data.Conduit.Classy
+                       Data.Conduit.Container
   build-depends:       base                          >= 4          && < 5
                      , conduit                       >= 0.5.4.1    && < 0.6
                      , xml-conduit                   >= 1.0        && < 1.1
-                     , classy-prelude                >= 0.4.1      && < 0.5
+                     , classy-prelude                >= 0.4.2      && < 0.5
+                     , transformers
+                     , monad-control
+                     , resourcet
+                     , void
+                     , bytestring
   ghc-options:         -Wall -fno-warn-orphans
+
+test-suite spec
+  type:           exitcode-stdio-1.0
+  main-is:        Spec.hs
+  other-modules:  Data.Conduit.ClassySpec
+                  Data.Conduit.ContainerSpec
+  hs-source-dirs: test
+  build-depends:  base
+                , hspec
+                , classy-prelude-conduit
+                , bytestring
+                , transformers
+                , QuickCheck
+                , conduit
 
 source-repository head
   type:     git
diff --git a/test/Data/Conduit/ClassySpec.hs b/test/Data/Conduit/ClassySpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Data/Conduit/ClassySpec.hs
@@ -0,0 +1,24 @@
+module Data.Conduit.ClassySpec where
+
+import Test.Hspec
+import Data.Conduit.Classy
+import qualified Data.Conduit.List as CL
+
+spec :: Spec
+spec = do
+    describe "connecting" $ do
+        it "works" $ do
+            let sink :: Int -> Sink Char IO Int
+                sink i = await >>= maybe (return i) (const $ sink $ i + 1)
+            let str = "hello world"
+            x <- mapM_ yield str $$ sink 0
+            x `shouldBe` length str
+    describe "connect-and-resume" $ do
+        it "works" $ do
+            let src :: Source IO Int
+                src = mapM_ yield [1..30]
+                take' = Sink . CL.take
+            (r1, x) <- src $$+ take' 10
+            (r2, y) <- r1 $$++ take' 10
+            z <- r2 $$+- Sink CL.consume
+            [x, y, z] `shouldBe` [[1..10], [11..20], [21..30]]
diff --git a/test/Data/Conduit/ContainerSpec.hs b/test/Data/Conduit/ContainerSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Data/Conduit/ContainerSpec.hs
@@ -0,0 +1,39 @@
+module Data.Conduit.ContainerSpec where
+
+import Test.Hspec
+import Test.Hspec.QuickCheck
+import Test.QuickCheck.Arbitrary
+import Data.Conduit.Classy
+import qualified Data.Conduit.Container as C
+import Data.Functor.Identity (Identity, runIdentity)
+import qualified Data.ByteString as S
+import qualified Data.ByteString.Lazy as L
+
+spec :: Spec
+spec = do
+    describe "Singleton" $ do
+        prop "consumes" $ \x ->
+            runIdentity ((C.toSource x :: Source Identity (C.Singleton Int)) $$ C.consume) == x
+        prop "takes" $ \str i' ->
+            let x = (C.toSource str :: Source Identity (C.Singleton Char)) $$ C.take i
+                i = abs i'
+             in runIdentity x == take i str
+    describe "ByteString" $ do
+        prop "consumes" $ \(ArbLByteString x) ->
+            runIdentity ((C.toSource x :: Source Identity S.ByteString) $$ C.consume) == x
+        prop "takes" $ \(ArbLByteString str) i' ->
+            let x = (C.toSource str :: Source Identity S.ByteString) $$ C.take i
+                i = abs i'
+             in runIdentity x == L.take (fromIntegral i) str
+
+newtype ArbByteString = ArbByteString { unArbByteString :: S.ByteString }
+    deriving Show
+
+instance Arbitrary ArbByteString where
+    arbitrary = fmap (ArbByteString . S.pack) arbitrary
+
+newtype ArbLByteString = ArbLByteString L.ByteString
+    deriving Show
+
+instance Arbitrary ArbLByteString where
+    arbitrary = fmap (ArbLByteString . L.fromChunks . map unArbByteString) arbitrary
diff --git a/test/Spec.hs b/test/Spec.hs
new file mode 100644
--- /dev/null
+++ b/test/Spec.hs
@@ -0,0 +1,1 @@
+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}