diff --git a/src/Node.Stream.CSV.Parse.purs b/src/Node.Stream.CSV.Parse.purs
index 80dd54c..1eb6f3a 100644
--- a/src/Node.Stream.CSV.Parse.purs
+++ b/src/Node.Stream.CSV.Parse.purs
@@ -87,6 +87,18 @@ parse config csv = do
   liftEffect $ Stream.end stream
   readAll stream
 
+-- | Loop until the stream is closed, invoking the callback with each record as it is parsed.
+foreach :: forall @r rl x. RowToList r rl => ReadCSVRecord r rl => CSVParser r x -> ({ | r } -> Aff Unit) -> Aff Unit
+foreach stream cb = whileJust do
+  isReadable <- liftEffect $ Stream.readable stream
+  when (not isReadable) $ makeAff \res -> mempty <* flip (Event.once Stream.readableH) stream $ res $ Right unit
+  whileJust do
+    r <- liftEffect $ read @r stream
+    for_ r cb
+    pure $ void r
+  isClosed <- liftEffect $ Stream.closed stream
+  pure $ if isClosed then Nothing else Just unit
+
 -- | Reads a parsed record from the stream.
 -- |
 -- | Returns `Nothing` when either:
@@ -101,18 +113,7 @@ read stream = runMaybeT do
 readAll :: forall @r rl a. RowToList r rl => ReadCSVRecord r rl => CSVParser r a -> Aff (Array { | r })
 readAll stream = do
   records <- liftEffect $ ST.toEffect $ Array.ST.new
-
-  whileJust do
-    isReadable <- liftEffect $ Stream.readable stream
-    when (not isReadable) $ makeAff \res -> mempty <* flip (Event.once Stream.readableH) stream $ res $ Right unit
-    liftEffect $ whileJust do
-      r <- read @r stream
-      for_ r \r' -> ST.toEffect $ Array.ST.push r' records
-      pure $ void r
-
-    isClosed <- liftEffect $ Stream.closed stream
-    pure $ if isClosed then Nothing else Just unit
-
+  foreach stream $ void <<< liftEffect <<< ST.toEffect <<< flip Array.ST.push records
   liftEffect $ ST.toEffect $ Array.ST.unsafeFreeze records
 
 -- | `data` event. Emitted when a CSV record has been parsed.
diff --git a/src/Node.Stream.CSV.Stringify.purs b/src/Node.Stream.CSV.Stringify.purs
index d4fda82..5f40b1c 100644
--- a/src/Node.Stream.CSV.Stringify.purs
+++ b/src/Node.Stream.CSV.Stringify.purs
@@ -79,22 +79,22 @@ stringify config records = do
 write :: forall @r rl a. RowToList r rl => WriteCSVRecord r rl => CSVStringifier r a -> { | r } -> Effect Unit
 write s = writeImpl s <<< writeCSVRecord @r @rl
 
+-- | Loop until the stream is closed, invoking the callback with each chunk of stringified CSV text.
+foreach :: forall r x. CSVStringifier r x -> (String -> Aff Unit) -> Aff Unit
+foreach stream cb = whileJust do
+  isReadable <- liftEffect $ Stream.readable stream
+  when (not isReadable) $ makeAff \res -> mempty <* flip (Event.once Stream.readableH) stream $ res $ Right unit
+  whileJust do
+    s <- liftEffect $ (join <<< map blush) <$> Stream.readEither stream
+    for_ s cb
+    pure $ void s
+  isClosed <- liftEffect $ Stream.closed stream
+  pure $ if isClosed then Nothing else Just unit
+
 -- | Read the stringified chunks until end-of-stream, returning the entire CSV string.
 readAll :: forall r a. CSVStringifier r a -> Aff String
 readAll stream = do
   chunks <- liftEffect $ ST.toEffect $ Array.ST.new
-
-  whileJust do
-    isReadable <- liftEffect $ Stream.readable stream
-    when (not isReadable) $ makeAff \res -> mempty <* flip (Event.on Stream.readableH) stream $ res $ Right unit
-
-    liftEffect $ whileJust do
-      s <- (join <<< map blush) <$> Stream.readEither stream
-      for_ s \s' -> ST.toEffect $ Array.ST.push s' chunks
-      pure $ void s
-
-    isClosed <- liftEffect $ Stream.closed stream
-    pure $ if isClosed then Nothing else Just unit
-
+  foreach stream $ void <<< liftEffect <<< ST.toEffect <<< flip Array.ST.push chunks
   chunks' <- liftEffect $ ST.toEffect $ Array.ST.unsafeFreeze chunks
   pure $ fold chunks'
diff --git a/src/Node.Stream.CSV.purs b/src/Node.Stream.CSV.purs
index b2550bc..d47d642 100644
--- a/src/Node.Stream.CSV.purs
+++ b/src/Node.Stream.CSV.purs
@@ -1,2 +1 @@
 module Node.Stream.CSV where
-