diff --git a/ChangeLog.md b/ChangeLog.md
new file mode 100644
--- /dev/null
+++ b/ChangeLog.md
@@ -0,0 +1,3 @@
+# Changelog for piped
+
+## Unreleased changes
diff --git a/LICENSE b/LICENSE
new file mode 100644
--- /dev/null
+++ b/LICENSE
@@ -0,0 +1,30 @@
+Copyright Author name here (c) 2019
+
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+    * Redistributions of source code must retain the above copyright
+      notice, this list of conditions and the following disclaimer.
+
+    * Redistributions in binary form must reproduce the above
+      copyright notice, this list of conditions and the following
+      disclaimer in the documentation and/or other materials provided
+      with the distribution.
+
+    * Neither the name of Author name here nor the names of other
+      contributors may be used to endorse or promote products derived
+      from this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
diff --git a/Setup.hs b/Setup.hs
new file mode 100644
--- /dev/null
+++ b/Setup.hs
@@ -0,0 +1,2 @@
+import Distribution.Simple
+main = defaultMain
diff --git a/piped.cabal b/piped.cabal
new file mode 100644
--- /dev/null
+++ b/piped.cabal
@@ -0,0 +1,106 @@
+cabal-version: 1.12
+
+-- This file has been generated from package.yaml by hpack version 0.31.1.
+--
+-- see: https://github.com/sol/hpack
+--
+-- hash: be93d4d46d6f3b7018cd97af12bc628bf189c2940205fea9d2bc5949051b0c11
+
+name:           piped
+version:        0.1.0.0
+synopsis:       Conduit with a smaller core
+description:    Please see the README on GitHub at <https://github.com/ssadler/piped#readme>
+category:       Control
+homepage:       https://github.com/ssadler/piped#readme
+bug-reports:    https://github.com/ssadler/piped/issues
+author:         Scott Sadler
+maintainer:     example@example.com
+copyright:      2019 Scott Sadler
+license:        BSD3
+license-file:   LICENSE
+build-type:     Simple
+extra-source-files:
+    ChangeLog.md
+
+source-repository head
+  type: git
+  location: https://github.com/ssadler/piped
+
+flag bench
+  description: Enable build benchmark
+  manual: True
+  default: False
+
+library
+  exposed-modules:
+      Piped
+      Piped.Compose
+      Piped.Extras
+      Piped.Internal
+      Piped.Prelude
+      Piped.Resume
+  other-modules:
+      Paths_piped
+  hs-source-dirs:
+      src
+  build-depends:
+      base >=4.7 && <5
+    , mtl
+  default-language: Haskell2010
+
+executable piped-bench
+  main-is: Bench.hs
+  other-modules:
+      BenchPipes
+      BenchYields
+      BenchCompare
+      Paths_piped
+  hs-source-dirs:
+      test-bench
+      test-common
+  default-extensions: LambdaCase FlexibleContexts
+  ghc-options: -rtsopts -O2 -fprof-auto -fwarn-unused-imports
+  build-depends:
+      base >=4.7 && <5
+    , bench-show
+    , conduit
+    , gauge
+    , mtl
+    , piped
+    , split
+  if (flag(bench))
+    buildable: True
+  else
+    buildable: False
+  default-language: Haskell2010
+
+test-suite piped-test
+  type: exitcode-stdio-1.0
+  main-is: Discover.hs
+  other-modules:
+      BenchSpec
+      ComposeSpec
+      ConduitSpec
+      PipeLike
+      TestUtils
+      BenchCompare
+      Paths_piped
+  hs-source-dirs:
+      test
+      test-common
+  default-extensions: RankNTypes MultiParamTypeClasses FunctionalDependencies FlexibleInstances ConstraintKinds TypeSynonymInstances AllowAmbiguousTypes LambdaCase FlexibleContexts
+  ghc-options: -threaded -rtsopts -with-rtsopts=-N -fwarn-unused-imports
+  build-depends:
+      base >=4.7 && <5
+    , conduit
+    , gauge
+    , microlens-platform
+    , mtl
+    , piped
+    , quickcheck-instances
+    , tasty
+    , tasty-discover
+    , tasty-golden
+    , tasty-hunit
+    , tasty-quickcheck
+  default-language: Haskell2010
diff --git a/src/Piped.hs b/src/Piped.hs
new file mode 100644
--- /dev/null
+++ b/src/Piped.hs
@@ -0,0 +1,31 @@
+
+module Piped (
+
+  -- Core
+  
+    Pipe
+  , await
+  , yield
+  , (.|)
+  , (|.)
+  , runPipe
+  , leftover
+
+  -- Barebones API
+  
+  , awaitForever
+  , awaitMaybe
+  , awaitJust
+  , yieldM
+  , sourceList
+  , sinkList
+  , sinkNull
+
+  , Void
+
+  ) where
+
+import Piped.Extras
+import Piped.Compose
+import Piped.Internal
+
diff --git a/src/Piped/Compose.hs b/src/Piped/Compose.hs
new file mode 100644
--- /dev/null
+++ b/src/Piped/Compose.hs
@@ -0,0 +1,100 @@
+
+-- | Pipe composition can be supply or demand driven, which refers to how the pipeline is
+--   initiated. Supply driven initiates the left side first.
+--
+--   When a pipe terminates, one side must return a value. Unless the pipeline is run
+--   in resumable mode, the other sides may never "complete" in that they do not
+--   return a value. If they allocate resources, MonadResource or similar should be
+--   used to handle cleanup (see 'Piped.Extras.bracketPipe').
+--
+--   Right hand termination works by indicating that no more values are available.
+--
+--   In a mode where either side can return a value, no special termination logic is invoked,
+--   execution ends the first time a side returns.
+--
+--   Left termination is not provided; since there is no way to indicate to the left side
+--   that the right has terminated, and potential solutions involve discarding values.
+--   However, the "either" modes are also suitable for left termination.
+--
+
+module Piped.Compose
+  (
+  -- ** Operators
+  --
+    (.|)
+  , (|.)
+
+  -- ** Demand driven
+  --
+  -- | The right hand side is run first. The left hand side is only
+  --   invoked by calling `await`.
+  --
+  , composeDemand
+  , composeDemandEither
+  
+  -- ** Supply driven
+  --
+  -- | The left hand side is run first. If it returns immediately, the right
+  --   hand side is invoked only in the case of `composeSupply`.
+  --   
+  , composeSupply
+  , composeSupplyEither
+  ) where
+
+import Piped.Internal
+
+
+-- | Demand driven; same as 'composeDemand
+--
+(.|) :: Monad m => Pipe i e m () -> Pipe e o m b -> Pipe i o m b
+(.|) = composeDemand
+{-# INLINE (.|) #-}
+
+
+-- | Supply driven; same as 'composeSupplyEither
+--
+(|.) :: Monad m => Pipe i e m a -> Pipe e o m a -> Pipe i o m a
+(|.) = composeSupplyEither
+{-# INLINE (|.) #-}
+
+
+-- | The right side is run first, only the right side may return a value.
+--
+composeDemand :: Monad m => Pipe i e m () -> Pipe e o m b -> Pipe i o m b
+composeDemand (Pipe f1) (Pipe f2) =
+  Pipe $ \rest l r ->
+    f2 (\_ -> rest termLeft) (Await $ f1 (\_ r _ -> terminate r) l) r
+{-# INLINE composeDemand #-}
+
+
+-- | The right side is run first, either side may return a value.
+--
+composeDemandEither :: Monad m => Pipe i e m a -> Pipe e o m a -> Pipe i o m a
+composeDemandEither (Pipe f1) (Pipe f2) =
+  Pipe $ \rest l r ->
+    f2 (\_ -> rest termLeft) (Await $ f1 (\l -> rest l . termRight) l) r
+{-# INLINE composeDemandEither #-}
+
+
+-- | The left side is run first, only the right side may return a value.
+--
+composeSupply :: Monad m => Pipe i e m () -> Pipe e o m b -> Pipe i o m b
+composeSupply (Pipe f1) (Pipe f2) =
+  Pipe $ \rest l r ->
+    f1 (\_ r _ -> terminate r) l $
+      Yield
+        (           f2 (\_ -> rest termLeft) termLeft             r)
+        (\i left -> f2 (\_ -> rest termLeft) (addLeftover i left) r)
+{-# INLINE composeSupply #-}
+
+
+-- | The left side is run first, either side may return a value.
+--
+composeSupplyEither :: Monad m => Pipe i e m a -> Pipe e o m a -> Pipe i o m a
+composeSupplyEither (Pipe f1) (Pipe f2) =
+  Pipe $ \rest l r ->
+    f1 (\l r -> rest l $ termRight r) l $
+      Yield
+        (           f2 (\_ -> rest termLeft) termLeft             r)
+        (\i left -> f2 (\_ -> rest termLeft) (addLeftover i left) r)
+{-# INLINE composeSupplyEither #-}
diff --git a/src/Piped/Extras.hs b/src/Piped/Extras.hs
new file mode 100644
--- /dev/null
+++ b/src/Piped/Extras.hs
@@ -0,0 +1,84 @@
+-- | Convenience methods, plus 'bracketPipe'.
+--
+
+module Piped.Extras where
+
+import Data.Function
+
+import Piped.Internal
+
+
+-- | Yield all values from a foldable data structure such as a list.
+--
+sourceList :: (Monad m, Foldable t) => t o -> Pipe () o m ()
+sourceList = foldMap yield
+{-# INLINE sourceList #-}
+
+-- | Return all input values as a list.
+--
+sinkList :: Pipe i Void m [i]
+sinkList = Pipe $
+  \rest l r ->
+    fix2 l id $
+      \go l fxs ->
+        let term = rest termLeft r $ fxs []
+         in runAwait l term $ \i l -> go l $ fxs . (i:)
+{-# INLINE sinkList #-}
+
+-- | Consume all values using given function
+--
+awaitForever :: Monad m => (i -> Pipe i o m a) -> Pipe i o m ()
+awaitForever f = Pipe $
+  \rest ->
+    fix $
+      \next await yield -> 
+        runAwait await (rest termLeft yield ()) $
+          \i await ->
+            unPipe (f i) (\l r _ -> next l r) await yield
+{-# INLINE awaitForever #-}
+
+-- | Discard all input values
+--
+sinkNull :: Monad m => Pipe i Void m ()
+sinkNull = Pipe $
+  \rest l r -> do
+    let term = rest termLeft r ()
+        f l = runAwait l term $ \_ -> f
+     in f l
+{-# INLINE sinkNull #-}
+
+-- | Basically `await >>= maybe ...`, but avoids the conversion to a maybe type.
+--
+awaitMaybe :: Pipe i o m a -> (i -> Pipe i o m a) -> Pipe i o m a
+awaitMaybe def act = Pipe $
+  \rest l r ->
+    runAwait l (unPipe def rest termLeft r)
+               (\i l -> unPipe (act i) rest l r)
+{-# INLINE awaitMaybe #-}
+
+
+-- | Act on a value, if there is no value then return ().
+--
+awaitJust :: Monad m => (i -> Pipe i o m ()) -> Pipe i o m ()
+awaitJust act = Pipe $
+  \rest l r ->
+    runAwait l (rest termLeft r ()) $
+      \i l -> unPipe (act i) rest l r
+{-# INLINE awaitJust #-}
+
+
+-- | Yield the result of a monadic action.
+--
+yieldM :: Monad m => Pipe i o m o -> Pipe i o m ()
+yieldM = (>>= yield)
+{-# INLINE yieldM #-}
+
+
+-- | This is the same as [bracketP](http://hackage.haskell.org/package/conduit-1.3.1.1/docs/Data-Conduit.html#v:bracketP) from Conduit, however it's not specialised to MonadResource.
+--
+bracketPipe :: Monad m => m b -> (b -> m ()) -> (b -> Pipe i o m a) -> Pipe i o m a
+bracketPipe alloc free inside = Pipe $
+  \rest l r -> do
+    b <- alloc
+    let rest' l r a = free b >> rest l r a
+    unPipe (inside b) rest' l r
diff --git a/src/Piped/Internal.hs b/src/Piped/Internal.hs
new file mode 100644
--- /dev/null
+++ b/src/Piped/Internal.hs
@@ -0,0 +1,189 @@
+{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE UndecidableInstances #-}
+
+-- | The internal API and core datatypes of Pipe.
+--
+module Piped.Internal
+  (
+  -- ** Pipe
+  --
+  -- | Piped is implemented using a type of codensity transform, that uses left and right continuations, rather than sum types, to propagate state transitions.
+  --
+    Pipe(..)
+  , await
+  , yield
+  , runPipe
+  , leftover
+
+  -- ** Continuations
+  --
+  -- | The state of a pipeline is encoded in a recursive data structure (Await / Yield)
+  --   which holds continuations that promise a final return value of @r@.
+  --
+  , Await(..)
+  , Yield(..)
+  , Await'
+  , Yield'
+  , runAwait
+  , runYield
+  , termLeft
+  , termRight
+  , voidRight
+  , addLeftover
+  , Void
+
+  -- ** Miscellania
+  --
+  , fix1
+  , fix2
+  ) where
+
+
+import Data.Void
+
+import Control.Monad
+import Control.Monad.Reader
+import Control.Monad.State
+
+
+-- | The upstream continuation; holds a callback that provides a value.
+--
+newtype Await i m a = Await { unAwait :: Await' i m a }
+
+
+-- | The downstream continuation; holds a callback that accepts a value,
+--   plus a default return value in case there is no more input.
+--
+data Yield i m a = Yield
+  { terminate :: m a
+  , unYield   :: Yield' i m a
+  }
+
+-- | The type of an upstream continuation
+--
+type Await' i m a = Yield i m a -> m a
+
+-- | The type of a downstream continuation
+--
+type Yield' i m a = i -> Await i m a -> m a
+
+-- | Run an Await
+--
+runAwait :: Await i m a -> m a -> Yield' i m a -> m a
+runAwait (Await awt) a yld = awt $ Yield a yld
+{-# INLINE runAwait #-}
+
+-- | Run a Yield
+--
+runYield :: Yield i m a -> i -> Await' i m a -> m a
+runYield (Yield _ a) i = a i . Await
+{-# INLINE runYield #-}
+
+-- | An Await that terminates when called.
+--
+termLeft :: Await i m a
+termLeft = Await terminate
+{-# INLINE termLeft #-}
+
+-- | A Yield that terminates when called.
+--
+termRight :: Yield i1 m a -> Yield i2 m a
+termRight r = Yield (terminate r) (\_ _ -> terminate r)
+{-# INLINE termRight #-}
+
+-- | A Void output
+--
+voidRight :: Yield Void m a
+voidRight = Yield (error "Void") (\i _ -> absurd i)
+{-# INLINE voidRight #-}
+
+-- | Wrap an Await with a leftover value
+--
+addLeftover :: i -> Await i m a -> Await i m a
+addLeftover i await = Await $ \y -> unYield y i await
+
+
+-- | The Pipe datatype, that represents a stage in a pipeline with inputs of type @i@ and outputs of type @o@.
+--
+newtype Pipe i o m a =
+  Pipe { unPipe :: forall r. (Await i m r -> Yield o m r -> a -> m r) -> Await i m r -> Yield o m r -> m r }
+
+instance Monad m => Functor (Pipe i o m) where
+  fmap f (Pipe p) = Pipe $
+    \rest l r -> p (\l r -> rest l r . f) l r
+
+instance Monad m => Applicative (Pipe i o m) where
+  pure = return
+  (<*>) = ap
+
+instance Monad m => Monad (Pipe i o m) where
+  return x = Pipe $ \f l r -> f l r x
+  Pipe f >>= g = Pipe $
+    \rest -> f (\l r a -> unPipe (g a) rest l r)
+
+instance MonadTrans (Pipe i o) where
+  lift mf = Pipe $ \f l r -> mf >>= f l r
+
+instance MonadIO m => MonadIO (Pipe i o m) where
+  liftIO = lift . liftIO
+
+instance MonadReader r m => MonadReader r (Pipe i o m) where
+  ask = lift ask
+  local f (Pipe p) = Pipe $ \rest l r -> local f $ p rest l r
+
+instance MonadState s m => MonadState s (Pipe i o m) where
+  get = lift get
+  put = lift . put
+
+instance Monad m => Semigroup (Pipe i o m a) where
+  (<>) = (>>)
+
+instance Monad m => Monoid (Pipe i o m ()) where
+  mempty = pure ()
+
+
+-- | Await a value. Nothing indicates that there are no more values.
+--
+await :: Monad m => Pipe i o m (Maybe i)
+await = Pipe $
+  \rest l r ->
+    let term = rest termLeft r Nothing
+     in runAwait l term $ \i l -> rest l r $ Just i
+{-# INLINE await #-}
+
+
+-- | Yield a value to downstream.
+--
+yield :: o -> Pipe i o m ()
+yield i = Pipe $
+  \rest a y -> runYield y i $ \y -> rest a y ()
+{-# INLINE yield #-}
+
+
+-- | Run pipe to completion.
+--
+runPipe :: Monad m => Pipe () Void m r -> m r
+runPipe pipe = unPipe pipe (\_ _ -> pure) termLeft voidRight
+{-# INLINE runPipe #-}
+
+
+-- | Push a value back into the incoming pipeline
+--
+leftover :: i -> Pipe i o m ()
+leftover i = Pipe $
+  \rest l r -> rest (addLeftover i l) r ()
+{-# INLINE leftover #-}
+
+
+-- `fix` providing one value.
+fix1 :: a -> ((a -> b) -> a -> b) -> b
+fix1 a f = fix f a
+{-# INLINE fix1 #-}
+
+
+-- `fix` providing two values.
+fix2 :: a -> b -> ((a -> b -> c) -> a -> b -> c) -> c
+fix2 a b f = fix f a b
+{-# INLINE fix2 #-}
diff --git a/src/Piped/Prelude.hs b/src/Piped/Prelude.hs
new file mode 100644
--- /dev/null
+++ b/src/Piped/Prelude.hs
@@ -0,0 +1,142 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE BangPatterns #-}
+
+-- | Basic functionality, closely mirroring the list / Foldable api from Prelude.
+--
+-- This module should be imported qualified i.e. `import Piped.Prelude as P`.
+--
+module Piped.Prelude where
+
+import Control.Monad
+import Control.Monad.Trans
+
+import Prelude hiding (foldl, scanl, mapM_, last, take, drop, zipWith)
+
+import Piped
+import Piped.Internal
+
+
+-- | Yield only elements satisfying a predicate.
+--
+filter :: Monad m => (i -> Bool) -> Pipe i i m ()
+filter f = awaitForever $ \i -> if f i then yield i else pure ()
+{-# INLINE filter #-}
+
+-- | Yield values while they satisfy a predicate, then return.
+--
+takeWhile :: Monad m => (i -> Bool) -> Pipe i i m ()
+takeWhile f = go where
+  go = awaitJust $ \i -> if f i then yield i >> go else leftover i
+{-# INLINE takeWhile #-}
+
+-- | Drop values while they satisfy a predicate, then return.
+--
+--   Note: this does not yield any values, and should be combined with (>>).
+--
+dropWhile :: Monad m => (i -> Bool) -> Pipe i o m ()
+dropWhile f = go where
+  go = awaitJust $ \i -> if f i then go else leftover i
+{-# INLINE dropWhile #-}
+
+-- | Equivalent to `map id`.
+--
+identity :: Monad m => Pipe i i m ()
+identity = awaitForever yield
+{-# INLINE identity #-}
+
+-- | `map` with an accumulator.
+--
+scanl :: forall i o m. (o -> i -> o) -> o -> Pipe i o m ()
+scanl f s = Pipe $
+  \rest ->
+    fix1 s $
+      \next !s l r ->
+        runYield r s $ \r -> 
+          runAwait l (rest l r ()) $
+            \i l -> next (f s i) l r
+{-# INLINE scanl #-}
+
+-- | Left fold over input values.
+--
+foldl :: Monad m => (a -> i -> a) -> a -> Pipe i o m a
+foldl f start = Pipe $
+  \rest ->
+    fix1 start $
+      \next !s l r ->
+        runAwait l (rest termLeft r s) $
+          \i l -> next (f s i) l r
+{-# INLINE foldl #-}
+
+-- | Drop n values.
+--
+--   Note: This will not yield values and should be combined with `>>`
+--
+drop :: Monad m => Int -> Pipe i o m ()
+drop 0 = pure ()
+drop n = awaitJust (\_ -> drop $ n-1)
+{-# INLINE drop #-}
+
+-- | Take n values.
+--
+take :: Monad m => Int -> Pipe i i m ()
+take 0 = pure ()
+take n = awaitJust $ \i -> yield i >> take (n-1)
+{-# INLINE take #-}
+
+-- | Map a pure function over input values.
+--
+map :: Monad m => (i -> o) -> Pipe i o m ()
+map f = awaitForever $ yield . f
+{-# INLINE map #-}
+
+-- | Map a monadic function over input values.
+--
+mapM :: Monad m => (i -> m o) -> Pipe i o m ()
+mapM f = awaitForever $ \i -> lift (f i) >>= yield
+{-# INLINE mapM #-}
+
+-- | Map a monadic function over input values but don't yield anything.
+--
+mapM_ :: Monad m => (i -> m ()) -> Pipe i () m ()
+mapM_ f = awaitForever $ lift . f
+{-# INLINE mapM_ #-}
+
+-- | Skip the first value
+--
+tail :: Monad m => Pipe i i m ()
+tail = await >> identity
+{-# INLINE tail #-}
+
+-- | Return the last value.
+--
+last :: Monad m => Pipe i o m (Maybe i)
+last = go Nothing where
+  go x = awaitMaybe (pure x) (go . Just)
+{-# INLINE last #-}
+
+-- | Zip two pipes together.
+--
+zip :: Monad m => Pipe () o m () -> Pipe () o' m () -> Pipe () (o, o') m ()
+zip = zipWith (,)
+{-# INLINE zip #-}
+
+-- | Zip two pipes together with a pure function.
+--
+zipWith :: Monad m => (o -> o' -> a) -> Pipe () o m () -> Pipe () o' m () -> Pipe () a m ()
+zipWith f (Pipe f1) (Pipe f2) = Pipe go where
+  go rest _ r = loop (side f1) (side f2) r
+    where
+      exit y = rest termLeft y ()
+      side f = Await $ \y -> f (\_ y () -> terminate y) termLeft y
+      loop a1 a2 yield =
+        runAwait a1 (exit yield) $ \i1 a1 ->
+          runAwait a2 (exit yield) $ \i2 a2 ->
+            runYield yield (f i1 i2) $ \yield ->
+              loop a1 a2 yield
+
+-- | Concatenate foldable inputs to a single stream of outputs.
+--
+concat :: (Foldable t, Monad m) => Pipe (t i) i m ()
+concat = awaitForever $ foldMap yield
+{-# INLINE concat #-}
diff --git a/src/Piped/Resume.hs b/src/Piped/Resume.hs
new file mode 100644
--- /dev/null
+++ b/src/Piped/Resume.hs
@@ -0,0 +1,62 @@
+
+module Piped.Resume
+  ( createResumableSource
+  , createResumableSink
+  , runResumableSource
+  , runResumableSink
+  , ResumableSource(..)
+  , ResumableSink(..)
+  , ResumableResult(..)
+  ) where
+
+
+import Piped.Internal
+
+
+-- | Either a resumable source or sink, plus the result of the pipe that finished.
+--
+data ResumableResult e m a b =
+    ResumeSource (ResumableSource e m a b) b
+  | ResumeSink   (ResumableSink   e m a b) a
+
+
+-- | A source that may be resumed
+--
+newtype ResumableSource o m a b = ResumableSource (Await o m (ResumableResult o m a b))
+
+
+-- | A sink that may be resumed
+--
+newtype ResumableSink   i m a b = ResumableSink   (Yield i m (ResumableResult i m a b))
+
+
+-- | Create a resumable source from a Pipe
+--
+createResumableSource :: Monad m => Pipe () o m a -> ResumableSource o m a b
+createResumableSource (Pipe f) = ResumableSource $
+  Await $ f (\l r a -> pure $ ResumeSink (ResumableSink r) a) termLeft
+
+
+-- | Create a resumable sink from a Pipe
+--
+createResumableSink :: Monad m => Pipe i Void m b -> ResumableSink i m a b
+createResumableSink (Pipe f) = ResumableSink $
+   Yield
+     (           f resumeLeft termLeft             voidRight)
+     (\i left -> f resumeLeft (addLeftover i left) voidRight)
+  where
+    resumeLeft l _ b  = pure $ ResumeSource (ResumableSource l) b
+
+
+-- | Run a resumable source
+--
+runResumableSource :: Monad m => ResumableSource i m a b -> Pipe i Void m b -> m (ResumableResult i m a b)
+runResumableSource (ResumableSource source) (Pipe f) =
+  f (\l _ b -> pure $ ResumeSource (ResumableSource l) b) source voidRight
+
+
+-- | Run a resumable sink
+--
+runResumableSink :: Monad m => Pipe () o m a -> ResumableSink o m a b -> m (ResumableResult o m a b)
+runResumableSink (Pipe f) (ResumableSink sink) =
+  f (\_ r a -> pure $ ResumeSink (ResumableSink r) a) termLeft sink
diff --git a/test-bench/Bench.hs b/test-bench/Bench.hs
new file mode 100644
--- /dev/null
+++ b/test-bench/Bench.hs
@@ -0,0 +1,52 @@
+
+module Main where
+
+import BenchShow
+import Gauge.Main
+import Control.Exception
+import Data.List (sortOn)
+import Data.List.Split
+import System.Environment (getArgs)
+
+import BenchYields
+import BenchPipes
+import BenchCompare
+
+
+main = do
+  args <- getArgs
+  if take 1 args == ["charts"]
+     then renderCharts $ args !! 1
+     else do
+       defaultMain
+        [ benchPipes
+        , benchYields
+        , benchCompare
+        ]
+
+
+renderCharts file = do
+  stdgraph "mixed_naive" $ drop 1
+  stdgraph "n_stages" $ drop 1
+  stdgraph "mixed_optimised" $ drop 1
+  
+  graph file "yielding_vs_monadic" $
+    defaultConfig
+      { classifyBenchmark = \name ->
+          case splitOn "/" name of
+            ["pipes", b, c] -> Just (c, b)
+            _               -> Nothing
+      }
+
+  where
+    stdgraph name f =
+      handle (\e -> print (e::SomeException)) $
+        graph file name $
+          defaultConfig
+            { classifyBenchmark = \s ->
+                case f (splitOn "/" s) of
+                  [a, b, c] | a == name -> Just (c, b)
+                  _                     -> Nothing
+            , title = Just name
+            , selectGroups = sortOn $ (/="piped") . fst
+            }
diff --git a/test-bench/BenchPipes.hs b/test-bench/BenchPipes.hs
new file mode 100644
--- /dev/null
+++ b/test-bench/BenchPipes.hs
@@ -0,0 +1,149 @@
+{-# LANGUAGE BangPatterns #-}
+
+module BenchPipes where
+
+import Data.Function (fix)
+
+import Control.Monad
+
+import Piped
+import Piped.Internal
+import Piped.Prelude as P
+
+import Gauge.Main
+
+import Prelude hiding (scanl, foldl, dropWhile, takeWhile)
+
+
+benchPipes = bgroup "pipes"
+  [ bgroup "sinkNull"  [ bench "monadic" $ whnfAppIO (runSink sinkNullM) 1000
+                       , bench "yielding" $ whnfAppIO (runSink sinkNull) 1000
+                       ]
+  , bgroup "sinkList"  [ bench "monadic" $ nfAppIO (runSink sinkListM) 1000
+                       , bench "yielding" $ nfAppIO (runSink sinkList) 1000
+                       ]
+
+  , bgroup "foldl"     [ bench "monadic" $ nfAppIO bFoldlM 10000
+                       , bench "yielding" $ nfAppIO bFoldlY 10000
+                       ]
+
+  , bgroup "scanl"     [ bench "monadic" $ nfAppIO bScanlM 10000
+                       , bench "yielding" $ nfAppIO bScanlY 10000
+                       ]
+
+  , bgroup "dropWhile" [ bench "monadic" $ nfAppIO (bDoWhile dropWhile) 10000
+                       , bench "yielding" $ nfAppIO (bDoWhile dropWhileY) 10000
+                       ]
+
+  , bgroup "takeWhile" [ bench "monadic" $ nfAppIO (bDoWhile takeWhile) 10000
+                       , bench "yielding" $ nfAppIO (bDoWhile takeWhileY) 10000
+                       ]
+  , bgroup "identity"  [ bench "monadic" $ nfAppIO (bIdentity identity) 1000
+                       , bench "yielding" $ nfAppIO (bIdentity identityY) 1000
+                       ]
+  -- , bgroup "maybes"    [ bench "awaitMaybe" $ nfAppIO (bMaybes $ awaitMaybe mempty) 10000
+  --                      , bench "awaitJust" $ nfAppIO (bMaybes $ awaitJust) 10000
+  --                      ]
+  , bgroup "map"       [ bench "monadic" $ nfAppIO (bSimple $ P.map (+1)) 10000
+                       , bench "yielding" $ nfAppIO (bSimple $ pmapMon (+1)) 10000
+                       ]
+  , bgroup "filter"    [ bench "monadic" $ nfAppIO (bSimple $ P.filter even) 10000
+                       , bench "yielding" $ nfAppIO (bSimple $ pFilter even) 10000
+                       ]
+  ]
+
+source :: Int -> Pipe () Int IO ()
+source n = Prelude.mapM_ yield [0..n]
+
+runSink :: Pipe Int Void IO a -> Int -> IO a
+runSink sink n =
+  runPipe $ source n .| sink
+
+sinkNullM :: Monad m => Pipe i Void m ()
+sinkNullM = awaitForever (\_ -> pure ())
+
+sinkListM :: Monad m => Pipe i Void m [i]
+sinkListM = let f = awaitMaybe (pure []) (\i -> (i:) <$> f) in f
+{-# INLINE sinkListM #-}
+
+bFoldl p n = runPipe $ source n .| (p (+) 0 >> pure ())
+bFoldlM n = runPipe $ source n .| (foldl' (+) 0 >> pure ())
+bFoldlY n = runPipe $ source n .| (foldl (+) 0 >> pure ())
+
+foldl' :: Monad m => (a -> i -> a) -> a -> Pipe i o m a
+foldl' f s =
+  let next !s = awaitMaybe (pure s) (next . f s) in next s
+{-# INLINE foldl' #-}
+-- foldl' :: Monad m => (a -> i -> a) -> a -> Pipe i o m a
+-- foldl' f start = Pipe $
+--   \rest ->
+--     fix1 start $
+--       \next !s l r ->
+--         runAwait l (rest termLeft r s) $
+--           \i l -> next (f s i) l r
+-- {-# INLINE foldl' #-}
+
+bScanlM = runSink $ scanlM (+) 0 .| sinkNull
+bScanlY = runSink $ scanl (+) 0 .| sinkNull
+
+scanlM :: Monad m => (a -> b -> a) -> a -> Pipe b a m ()
+scanlM f =
+  fix $ \next !s ->
+    yield s >> awaitJust (\i -> next $ f s i)
+{-# INLINE scanlM #-}
+
+bDoWhile p n = runPipe $ source n .| p (\_ -> True) .| sinkNull
+
+dropWhileY :: Monad m => (i -> Bool) -> Pipe i i m ()
+dropWhileY f = Pipe $
+  \rest ->
+    fix $ \next l r ->
+      runAwait l (rest termLeft r ()) $ \i l ->
+        if f i
+           then next l r
+           else let l' = addLeftover i l
+                 in unPipe identity rest l' r
+{-# INLINE dropWhileY #-}
+
+takeWhileY :: Monad m => (i -> Bool) -> Pipe i i m ()
+takeWhileY f = Pipe $
+  \rest ->
+    fix $ \next l r ->
+      runAwait l (rest termLeft r ()) $ \i l ->
+        if f i
+           then runYield r i $ next l
+           else rest l r ()
+{-# INLINE takeWhileY #-}
+
+bIdentity :: Pipe Int Int IO () -> Int -> IO ()
+bIdentity p n = runPipe $ source n .| ((iterate (.| p) p) !! 10) .| sinkNull
+
+identityY :: Pipe i i m ()
+identityY = Pipe $
+  \rest ->
+    fix $ \next l r ->
+      runAwait l (rest termLeft r ()) $ \i l ->
+        runYield r i $ next l
+{-# INLINE identityY #-}
+
+bMaybes p n = runPipe $ source n .| go
+  where go = p $ (\_ -> go :: Pipe Int Void IO ())
+
+bSimple p n = runPipe $ source n .| p .| sinkNull
+
+pFilter :: (i -> Bool) -> Pipe i i m ()
+pFilter f = Pipe $
+  \rest -> fix $
+    \next l r ->
+      runAwait l (rest termLeft r ()) $
+        \i l -> if f i
+                   then runYield r i $ \r -> next l r
+                   else next l r
+{-# INLINE pFilter #-}
+
+pmapMon :: (i -> o) -> Pipe i o m ()
+pmapMon f = Pipe $
+  \rest -> fix $
+    \next l r ->
+      runAwait l (rest termLeft r ()) $
+        \i l -> runYield r (f i) $ \r -> next l r
diff --git a/test-bench/BenchYields.hs b/test-bench/BenchYields.hs
new file mode 100644
--- /dev/null
+++ b/test-bench/BenchYields.hs
@@ -0,0 +1,114 @@
+
+module BenchYields where
+
+import Gauge.Main
+
+-- Benchmark different data structures for continuation passing
+--
+-- The requirement is for two coroutines to pass continuations
+-- back and forth, so a recursive data structure is required.
+-- It needs to be able to pass a value as well as the
+-- contination in one direction, and indicate that no further
+-- value is available in the other.
+--
+-- Some results:
+--
+-- cont/maybe                               time                 16.66 μs
+-- cont/term                                time                 14.80 μs
+-- cont/stall                               time                 13.88 μs
+-- cont/term4                               time                 21.43 μs
+
+
+benchYields =
+  bgroup "cont" [ bench "maybe"  $ whnf runMaybes 1000
+                , bench "term"   $ whnf runTerm 1000
+                , bench "stall"  $ whnf runStall 1000
+                , bench "term4"  $ whnf runTerm4 1000
+                ]
+
+
+-- The first data structure wraps the data in a Maybe in order to indicate
+-- when no more data is available.
+
+newtype Await i a = Await { unAwait :: Yield i a -> a }
+newtype Yield i a = Yield { unYield :: Maybe i -> Await i a -> a }
+
+runMaybes n = unAwait (left n) right
+  where
+    left n =
+      Await $ \yield -> do
+        let v = if n == 0 then Nothing else Just ()
+         in unYield yield v $ left (n-1)
+      
+    right = Yield $ \mi await -> maybe () (\_ -> unAwait await right) mi
+
+
+--------------------------
+
+-- This is the preferred version; the yield function wrapper also
+-- contains a thunk which returns an output directly. It's faster
+-- than the above and can also express the Maybe interface.
+
+newtype Await2 i a = Await2 { unAwait2 :: Await2' i a }
+
+data Yield2 i a = Yield2
+  { terminate :: a
+  , unYield2 :: Yield2' i a
+  }
+
+type Await2' i a = Yield2 i a -> a
+type Yield2' i a = i -> Await2 i a -> a
+
+runTerm n = unAwait2 (left n) right
+  where
+    left n =
+      Await2 $ \yield ->
+        if n == 0 then terminate yield
+                  else unYield2 yield () $ left (n-1)
+      
+    right = Yield2 () $ \() await -> unAwait2 await right
+
+
+--------------------------
+
+-- This is slightly faster than the previous one, but since it does not allow
+-- to specify termination, any node can terminate and that's not always desirable.
+
+newtype Await3 i a = Await3 { unAwait3 :: Yield3 i a -> a }
+newtype Yield3 i a = Yield3 { unYield3 :: i -> Await3 i a -> a }
+
+runStall n = unAwait3 (left n) right
+  where
+    left n =
+      Await3 $ \yield -> do
+        if n == 0 then ()
+                  else unYield3 yield () $ left (n-1)
+      
+    right = Yield3 $ \() await -> unAwait3 await right
+
+
+--------------------------
+
+-- This one is just slow, its the same concept as Yield2 but uses an extra 
+-- function to select between yielding and termination, to avoid the `data`
+-- record, but actually the `data` record hardly incurs any penalty as it turns out.
+
+newtype Await4 i a = Await4 { unAwait4 :: Await4' i a }
+
+newtype Yield4 i a = Yield4
+  { unYield4 :: Yield4' i a
+  }
+
+type Await4' i a = Yield4 i a -> a
+type Yield4' i a = (a -> (i -> Await4 i a -> a) -> a) -> a
+
+runTerm4 n = unAwait4 (left n) right
+  where
+    left n =
+      Await4 $ \yield ->
+        unYield4 yield $
+          \term onval -> 
+            if n == 0 then term else onval () (left (n-1))
+      
+    right = Yield4 $ \f -> f () $ \() await -> unAwait4 await right
+
diff --git a/test-common/BenchCompare.hs b/test-common/BenchCompare.hs
new file mode 100644
--- /dev/null
+++ b/test-common/BenchCompare.hs
@@ -0,0 +1,150 @@
+{-# LANGUAGE AllowAmbiguousTypes #-}
+{-# LANGUAGE BangPatterns #-}
+
+module BenchCompare where
+
+import Control.Monad
+import Data.Function
+import Data.Void
+import Prelude hiding (map, foldl, takeWhile, dropWhile, scanl, filter)
+
+import Gauge.Main
+
+import qualified Conduit as C
+import qualified Data.Conduit.Combinators as C
+import qualified Data.Conduit.List as CL
+import qualified Piped as P
+import qualified Piped.Prelude as P
+
+
+benchCompare = bgroup "compare"
+  [ bgroup "n_stages" $
+      [ compare (show (n+2), p_stages n, c_stages n) | n <- [0, 2, 4, 6]
+      ]
+  , bgroup  "mixed_optimised" $
+      [ compare ("map",           p_map,           c_map)
+      , compare ("foldl",         p_foldl,         c_foldl)
+      , compare ("takeall", p_takeall, c_takeall)
+      , compare ("dropall", p_dropall, c_dropall)
+      , compare ("scanl",   p_scanl,   c_scanl)
+      , compare ("filter_all",    p_filter_all,    c_filter_all)
+      , compare ("filter_none",   p_filter_none,   c_filter_none)
+      , compare ("sinklist",      p_sinklist,      c_sinklist)
+      , compare ("sinknull",      p_sinknull,      c_sinknull)
+      ]
+  , bgroup "mixed_naive" $
+      [ compare ("map",           p_n_map,           c_n_map)
+      , compare ("foldl",         p_n_foldl,         c_n_foldl)
+      , compare ("takeall", p_n_takeall, c_n_takeall)
+      , compare ("dropall", p_n_dropall, c_n_dropall)
+      , compare ("scanl",   p_n_scanl,   c_n_scanl)
+      , compare ("filter_all",    p_n_filter_all,    c_n_filter_all)
+      , compare ("filter_none",   p_n_filter_none,   c_n_filter_none)
+      , compare ("sinklist",      p_n_sinklist,      c_n_sinklist)
+      , compare ("sinknull",      p_n_sinknull,      c_n_sinknull)
+      ]
+  ]
+
+  where
+    iter = 100000
+    runC :: (Integer -> C.ConduitT () o IO a) -> Integer -> IO ()
+    runC c n = C.runConduit $ (c n >> pure ()) C..| C.sinkNull
+    runP :: (Integer -> P.Pipe () o IO a) -> Integer -> IO ()
+    runP p n = P.runPipe    $ (p n >> pure ()) P..| P.sinkNull
+    compare :: (String, Integer -> P.Pipe () o IO a, Integer -> C.ConduitT () o IO a) -> Benchmark
+    compare (name, p, c) = bgroup name $
+      [ bench "conduit" $ whnfAppIO (runC c) iter
+      , bench "piped"   $ whnfAppIO (runP p) iter
+      ]
+
+
+c_stages s n = source_C n C..| go s C..| C.sinkNull where
+  go 0 = ident_C
+  go s = ident_C C..| go (s-1)
+
+p_stages s n = source_P n P..| go s P..| P.sinkNull where
+  go 0 = ident_P
+  go s = ident_P P..| go (s-1)
+
+
+c_foldl, c_n_foldl :: Integer -> C.ConduitT () () IO Integer
+p_foldl, p_n_foldl :: Integer -> P.Pipe () () IO Integer
+c_sinklist, c_n_sinklist :: Integer -> C.ConduitT () Void IO [Integer]
+p_sinklist, p_n_sinklist :: Integer -> P.Pipe () Void IO [Integer]
+
+-- Optimised
+
+c_map           n = CL.sourceList [0..n] C..| C.map (+1)
+c_foldl         n = CL.sourceList [0..n] C..| C.foldl (+) 0
+c_takeall       n = CL.sourceList [0..n] C..| C.takeWhile (<n)
+c_dropall       n = CL.sourceList [0..n] C..| (C.dropWhile (<n) >> ident_C)
+c_scanl         n = CL.sourceList [0..n] C..| C.scanl (+) 0
+c_filter_all    n = CL.sourceList [0..n] C..| C.filter (<0)
+c_filter_none   n = CL.sourceList [0..n] C..| C.filter (>=0)
+c_sinklist      n = CL.sourceList [0..quot n 2] C..| C.sinkList
+c_sinknull      n = CL.sourceList [0..n]
+
+p_map           n = P.sourceList [0..n] P..| P.map (+1)
+p_foldl         n = P.sourceList [0..n] P..| P.foldl (+) 0
+p_takeall       n = P.sourceList [0..n] P..| P.takeWhile (<n)
+p_dropall       n = P.sourceList [0..n] P..| (P.dropWhile (<n) >> ident_P)
+p_scanl         n = P.sourceList [0..n] P..| P.scanl (+) 0
+p_filter_all    n = P.sourceList [0..n] P..| P.filter (<0)
+p_filter_none   n = P.sourceList [0..n] P..| P.filter (>=0)
+p_sinklist      n = P.sourceList [0..quot n 2] P..| P.sinkList
+p_sinknull      n = P.sourceList [0..n]
+
+-- Naive
+
+c_n_map           n = source_C n C..| map_C (+1)
+c_n_foldl         n = source_C n C..| foldl_C (+) 0
+c_n_takeall       n = source_C n C..| takeWhile_C (<n)
+c_n_dropall       n = source_C n C..| (dropWhile_C (<n) >> ident_C)
+c_n_scanl         n = source_C n C..| scanl_C (+) 0
+c_n_filter_all    n = source_C n C..| filter_C (<0)
+c_n_filter_none   n = source_C n C..| filter_C (>=0)
+c_n_sinklist      n = source_C (quot n 2) C..| sinkList_C
+c_n_sinknull      n = source_C n
+
+p_n_map           n = source_P n P..| map_P (+1)
+p_n_foldl         n = source_P n P..| foldl_P (+) 0
+p_n_takeall       n = source_P n P..| takeWhile_P (<n)
+p_n_dropall       n = source_P n P..| (dropWhile_P (<n) >> ident_P)
+p_n_scanl         n = source_P n P..| scanl_P (+) 0
+p_n_filter_all    n = source_P n P..| filter_P (<0)
+p_n_filter_none   n = source_P n P..| filter_P (>=0)
+p_n_sinklist      n = source_P (quot n 2) P..| sinkList_P
+p_n_sinknull      n = source_P n
+
+
+
+ident_C = C.awaitForever C.yield
+source_C n = CL.sourceList [0..n]
+map_C f = C.awaitForever $ C.yield . f
+awaitMaybe_C d f = C.await >>= maybe d f
+foldl_C f = go where
+  go (!s) = awaitMaybe_C (pure s) (\i -> go (f s i))
+takeWhile_C f = go where
+  go = awaitMaybe_C (pure ()) (\i -> if f i then C.yield i >> go else pure ())
+dropWhile_C f = go where
+  go = awaitMaybe_C (pure ()) (\i -> if f i then go else C.yield i >> ident_C)
+scanl_C f = go where
+  go (!s) = C.yield s >> awaitMaybe_C (pure ()) (\i -> go $ f s i)
+filter_C f = C.awaitForever $ \i -> if f i then C.yield i else pure ()
+sinkList_C = awaitMaybe_C (pure []) (\i -> (i:) <$> sinkList_C)
+
+
+ident_P = P.awaitForever P.yield
+source_P n = P.sourceList [0..n]
+map_P f = P.awaitForever $ P.yield . f
+awaitMaybe_P d f = P.await >>= maybe d f
+foldl_P f = go where
+  go (!s) = awaitMaybe_P (pure s) (\i -> go (f s i))
+takeWhile_P f = go where
+  go = awaitMaybe_P (pure ()) (\i -> if f i then P.yield i >> go else pure ())
+dropWhile_P f = go where
+  go = awaitMaybe_P (pure ()) (\i -> if f i then go else P.yield i >> ident_P)
+scanl_P f = go where
+  go (!s) = P.yield s >> awaitMaybe_P (pure ()) (\i -> go $ f s i)
+filter_P f = P.awaitForever $ \i -> if f i then P.yield i else pure ()
+sinkList_P = awaitMaybe_P (pure []) (\i -> (i:) <$> sinkList_P)
diff --git a/test/BenchSpec.hs b/test/BenchSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/BenchSpec.hs
@@ -0,0 +1,49 @@
+
+module BenchSpec where
+
+import Data.Conduit as C
+import Data.Conduit.Combinators as CL
+import Piped as P
+
+import Test.Tasty
+import Test.Tasty.HUnit
+
+import BenchCompare
+
+
+test_bench :: TestTree
+test_bench = testGroup "Verify that benchmark pipes output same results" $
+  [ testGroup "n_stages" $
+      [ -- compare (show (n+2), p_stages n, c_stages n) | n <- [0, 2, 4, 6]
+      ]
+  , testGroup  "mixed_optimised" $
+      [ compare ("map",           p_map,           c_map)
+      , compare ("foldl",         p_foldl,         c_foldl)
+      , compare ("takeall", p_takeall, c_takeall)
+      , compare ("dropall", p_dropall, c_dropall)
+      , compare ("scanl",   p_scanl,   c_scanl)
+      , compare ("filter_all",    p_filter_all,    c_filter_all)
+      , compare ("filter_none",   p_filter_none,   c_filter_none)
+      , compare ("sinklist",      p_sinklist,      c_sinklist)
+      , compare ("sinknull",      p_sinknull,      c_sinknull)
+      ]
+  , testGroup "mixed_naive" $
+      [ compare ("map",           p_n_map,           c_n_map)
+      , compare ("foldl",         p_n_foldl,         c_n_foldl)
+      , compare ("takeall", p_n_takeall, c_n_takeall)
+      , compare ("dropall", p_n_dropall, c_n_dropall)
+      , compare ("scanl",   p_n_scanl,   c_n_scanl)
+      , compare ("filter_all",    p_n_filter_all,    c_n_filter_all)
+      , compare ("filter_none",   p_n_filter_none,   c_n_filter_none)
+      , compare ("sinklist",      p_n_sinklist,      c_n_sinklist)
+      , compare ("sinknull",      p_n_sinknull,      c_n_sinknull)
+      ]
+  ]
+  where
+    n = 100
+
+    compare :: (Eq o, Show o) => (String, Integer -> Pipe () o IO a, Integer -> ConduitT () o IO a) -> TestTree
+    compare (s, p, c) = testCase s $ do
+      cr <- runConduit ((c n >> pure ()) C..| CL.sinkList)
+      pr <- runPipe    ((p n >> pure ()) P..| P.sinkList)
+      pr @?= cr
diff --git a/test/ComposeSpec.hs b/test/ComposeSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/ComposeSpec.hs
@@ -0,0 +1,31 @@
+
+module ComposeSpec where
+
+import Piped
+import Piped.Compose
+
+import TestUtils
+
+
+test_4 = testCase "composeSupply init correct order" $ do
+  let o = runPipeS [] $ composeSupply (alloc 1 >> yield 1) (alloc 2 >> await)
+  o @?= (Just 1, [1, 2])
+
+test_5 = testCase "composeSupplyEither left run first" $ do
+  let o = runPipeI $ composeSupplyEither (pure 1) (pure 2)
+  o @?= 1
+
+test_55 = testCase "composeSupplyEither right can return" $ do
+  let o = runPipeI $ composeSupplyEither (yield () >> pure 1) (pure (2::Int))
+  o @?= 2
+
+test_6 = testCase "composeDemand left doesnt run" $
+  runPipe $ composeDemand (error "failure") (pure ())
+
+test_8 = testCase "composeDemandEither right run first" $ do
+  let o = runPipeI $ composeDemandEither (pure 1) (pure 2)
+  o @?= 2
+
+test_88 = testCase "composeDemandEither left can return" $ do
+  let o = runPipeI $ composeDemandEither (pure 1) (await >> pure 2)
+  o @?= 1
diff --git a/test/ConduitSpec.hs b/test/ConduitSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/ConduitSpec.hs
@@ -0,0 +1,54 @@
+
+module ConduitSpec where
+
+import qualified Data.Conduit as C
+import qualified Piped as P
+import Control.Monad.State
+
+import Test.Tasty.HUnit
+
+import PipeLike as PL
+
+
+-- | Below tests polymorphic pipeline definitions using Piped and Conduit.
+--   The application of () is because for some reason GHC will allow 
+--   polymorphic definitions without signatures if an argument is passed,
+--   event a useless one.
+
+test_1 = test "demand driven" p p
+  where
+    p _ = undefined .| (pure ())
+
+test_2 = test "sinkList correct order" p p
+  where
+    p _ = mapM_ yield [True, False] .| sinkList
+
+test_3 = test "dropWhile doesn't consume" p p
+  where
+    p _ = sourceList [1..10] .| (PL.dropWhile (<5) >> leftover (-1) >> sinkList)
+
+test_4 = test "takeWhile doesn't drop values" p p
+  where
+    p _ = sourceList [1..10] .| (PL.takeWhile (<5) >> PL.map (+1)) .| sinkList
+
+test_44 = test "take doesn't drop values" p p
+  where
+    p _ = sourceList [1..10] .| PL.take 5 .| sinkList
+
+test_45 = test "drop doesn't yield values" p p
+  where
+    p _ = sourceList [1..10] .| (PL.drop 5 >> PL.map (+1)) .| sinkList
+
+test_5 = test "recursive pipes don't do funny things" p p
+  where
+    p _ = go 2
+    go 0 = await >> pure True
+    go n = yield () .| (await >> go (n-1))
+
+test_6 = test "foldl works as expected" p p
+  where
+    p _ = sourceList [True, False, True] .| PL.foldl (flip (:)) []
+
+
+test s p c = testCompare s (p ()) (c ())
+testCompare s c p = testCase s $ join $ (@?=) <$> P.runPipe p <*> C.runConduit c
diff --git a/test/Discover.hs b/test/Discover.hs
new file mode 100644
--- /dev/null
+++ b/test/Discover.hs
@@ -0,0 +1,1 @@
+{-# OPTIONS_GHC -F -pgmF tasty-discover -optF --tree-display #-}
diff --git a/test/PipeLike.hs b/test/PipeLike.hs
new file mode 100644
--- /dev/null
+++ b/test/PipeLike.hs
@@ -0,0 +1,87 @@
+{-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE FunctionalDependencies #-}
+{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE ConstraintKinds #-}
+{-# LANGUAGE TypeSynonymInstances #-}
+{-# LANGUAGE AllowAmbiguousTypes #-}
+{-# LANGUAGE LambdaCase #-}
+
+module PipeLike
+  ( module PipeLike
+  , C.ConduitT
+  , P.Pipe
+  , C.runConduit
+  , P.runPipe
+  , Void
+  ) where
+
+import qualified Data.Conduit as C
+import qualified Data.Conduit.Combinators as C
+import qualified Data.Conduit.List as C (sourceList)
+import qualified Piped as P
+import qualified Piped.Prelude as P
+
+import Data.Void
+
+
+class (Monad m, Monad (p i o m)) => PipeLike p i o m where
+  await :: p i o m (Maybe i)
+  yield :: o -> p i o m ()
+  (.|) :: p i e m () -> p e o m b -> p i o m b
+  infixr 5 .|
+  sourceList :: [o] -> p () o m ()
+  sinkList :: p i Void m [i]
+  foldl :: (a -> i -> a) -> a -> p i o m a
+  scanl :: (a -> b -> a) -> a -> p b a m ()
+  sinkNull :: p i Void m ()
+  awaitForever :: (i -> p i o m ()) -> p i o m ()
+  map :: (a -> b) -> p a b m ()
+  mapMC :: (i -> m o) -> p i o m ()
+  mapMC_ :: (i -> m ()) -> p i () m ()
+  take :: Int -> p i i m ()
+  drop :: Int -> p i o m ()
+  takeWhile :: (i -> Bool) -> p i i m ()
+  dropWhile :: Monad m => (i -> Bool) -> p i o m ()
+  filter :: (i -> Bool) -> p i i m ()
+  leftover :: i -> p i o m ()
+
+instance Monad m => PipeLike C.ConduitT i o m where
+  await = C.await
+  yield = C.yield
+  (.|) = (C..|)
+  sourceList = C.sourceList
+  sinkList = C.sinkList
+  foldl = C.foldl
+  scanl = C.scanl
+  sinkNull = C.sinkNull
+  awaitForever = C.awaitForever
+  take = C.take
+  drop = C.drop
+  takeWhile = C.takeWhile
+  map = C.map
+  mapMC = C.mapM
+  mapMC_ = C.mapM_
+  dropWhile = C.dropWhile
+  filter = C.filter
+  leftover = C.leftover
+
+instance Monad m => PipeLike P.Pipe i o m where
+  await = P.await
+  yield = P.yield
+  (.|) = (P..|)
+  sourceList = P.sourceList
+  sinkList = P.sinkList
+  foldl = P.foldl
+  scanl = P.scanl
+  sinkNull = P.sinkNull
+  awaitForever = P.awaitForever
+  take = P.take
+  drop = P.drop
+  takeWhile = P.takeWhile
+  map = P.map
+  mapMC = P.mapM
+  mapMC_ = P.mapM_
+  dropWhile = P.dropWhile
+  filter = P.filter
+  leftover = P.leftover
diff --git a/test/TestUtils.hs b/test/TestUtils.hs
new file mode 100644
--- /dev/null
+++ b/test/TestUtils.hs
@@ -0,0 +1,21 @@
+
+module TestUtils
+  ( module TestUtils
+  , module T
+  ) where
+
+import Data.IORef as T
+import Test.Tasty as T
+import Test.Tasty.HUnit as T
+
+import Control.Monad.State as T
+import Control.Monad.Identity as T
+
+import Piped as T
+import Piped.Internal as T
+
+
+runPipeI = runIdentity . runPipe
+runPipeS s p = runState (runPipe p) s
+alloc n = modify (++[n])
+