diff --git a/dataflower.cabal b/dataflower.cabal
--- a/dataflower.cabal
+++ b/dataflower.cabal
@@ -4,10 +4,10 @@
 --
 -- see: https://github.com/sol/hpack
 --
--- hash: f3c31e7bf792a3456fb07d2fe54ce9b968e689a915ac7ce9b658e8c99697b860
+-- hash: d435fe0abed8e5cf0d8dde80a09574abd559ab455d29bc746af5cf84894e06c8
 
 name:           dataflower
-version:        0.2.1.0
+version:        0.2.2.0
 synopsis:       A Pure-Haskell Timely Dataflow System
 description:    See README
 homepage:       https://github.com/doublecrowngaming/dataflower#readme
diff --git a/src/Test/Dataflow.hs b/src/Test/Dataflow.hs
--- a/src/Test/Dataflow.hs
+++ b/src/Test/Dataflow.hs
@@ -1,23 +1,49 @@
 module Test.Dataflow (
-  runDataflow
+  runDataflow,
+  runDataflowMany
 ) where
 
-import           Control.Concurrent.STM.TVar (newTVarIO, readTVarIO)
-import           Control.Monad               (void)
+import           Control.Concurrent.STM.TVar (modifyTVar', newTVarIO,
+                                              readTVarIO)
+import           Control.Monad               (foldM_)
 import           Control.Monad.IO.Class      (MonadIO (..))
-import           Dataflow                    (Dataflow, Edge, compile, execute,
-                                              outputTVar)
+import           Control.Monad.STM           (atomically)
+import           Control.Monad.Trans.Class   (lift)
+import           Dataflow                    (Edge, compile, execute,
+                                              modifyState, readState,
+                                              statefulVertex, writeState)
+import           Dataflow.Primitives         (Dataflow (Dataflow))
 import           Prelude
 
--- | Run a dataflow with a list of inputs.
+-- | Run a dataflow with a list of inputs. All inputs will be sent as part of
+-- a single epoch.
 --
 -- @since 0.1.0.0
 runDataflow :: MonadIO io => (Edge o -> Dataflow (Edge i)) -> [i] -> io [o]
-runDataflow dataflow inputs =
+runDataflow dataflow inputs = head <$> runDataflowMany dataflow [inputs]
+
+-- | Run a dataflow with a list of lists of inputs. Each outer list will be
+-- sent as its own epoch.
+--
+-- @since 0.2.2.0
+runDataflowMany :: MonadIO io => (Edge o -> Dataflow (Edge i)) -> [[i]] -> io [[o]]
+runDataflowMany dataflow inputs =
   liftIO $ do
     out     <- newTVarIO []
-    program <- compile (dataflow =<< outputTVar (:) out)
+    program <- compile (dataflow =<< outputTVarNestedList out)
 
-    void $ execute inputs program
+    foldM_ (flip execute) program inputs
 
     reverse <$> readTVarIO out
+
+  where
+    outputTVarNestedList register =
+      statefulVertex []
+        (\sref _ x -> modifyState sref (x :))
+        (\sref _   -> do
+          state <- readState sref
+
+          Dataflow $ lift $ atomically $ modifyTVar' register (reverse state :)
+
+          writeState sref []
+        )