chore: prepare v1.2.3

User-provided callback no longer blocks subsequent reads

bun/fmt.js

@@ -1,27 +1,27 @@
 /** @type {(parser: string, ps: string[]) => import("bun").Subprocess} */
 const prettier = (parser, ps) =>
-  Bun.spawn(['bun', 'x', 'prettier', '--write', '--parser', parser, ...ps], {
-    stdout: 'inherit',
-    stderr: 'inherit',
-  })
+  Bun.spawn(["bun", "x", "prettier", "--write", "--parser", parser, ...ps], {
+    stdout: "inherit",
+    stderr: "inherit",
+  });
 
 const procs = [
-  prettier('babel', ['./src/**/*.js', './bun/**/*.js', './.prettierrc.cjs']),
-  prettier('json', ['./package.json', './jsconfig.json']),
+  prettier("babel", ["./src/**/*.js", "./bun/**/*.js", "./.prettierrc.cjs"]),
+  prettier("json", ["./package.json", "./jsconfig.json"]),
   Bun.spawn(
     [
-      'bun',
-      'x',
-      'purs-tidy',
-      'format-in-place',
-      'src/**/*.purs',
-      'test/**/*.purs',
+      "bun",
+      "x",
+      "purs-tidy",
+      "format-in-place",
+      "src/**/*.purs",
+      "test/**/*.purs",
     ],
     {
-      stdout: 'inherit',
-      stderr: 'inherit',
+      stdout: "inherit",
+      stderr: "inherit",
     },
   ),
-]
+];
 
-await Promise.all(procs.map(p => p.exited))
+await Promise.all(procs.map((p) => p.exited));

bun/prepare.js

@@ -1,34 +1,34 @@
-import { readFile, writeFile } from 'fs/promises'
-import { execSync } from 'child_process'
+import { readFile, writeFile } from "fs/promises";
+import { execSync } from "child_process";
 
-let ver = process.argv[2]
+let ver = process.argv[2];
 if (!ver) {
-  console.error(`tag required: bun bun/prepare.js v1.0.0`)
-  process.exit(1)
+  console.error(`tag required: bun bun/prepare.js v1.0.0`);
+  process.exit(1);
 } else if (!/v\d+\.\d+\.\d+/.test(ver)) {
-  console.error(`invalid tag: ${ver}`)
-  process.exit(1)
+  console.error(`invalid tag: ${ver}`);
+  process.exit(1);
 }
 
-ver = (/\d+\.\d+\.\d+/.exec(ver) || [])[0] || ''
+ver = (/\d+\.\d+\.\d+/.exec(ver) || [])[0] || "";
 
-const pkg = await readFile('./package.json', 'utf8')
-const pkgnew = pkg.replace(/"version": ".+"/, `"version": "v${ver}"`)
-await writeFile('./package.json', pkgnew)
+const pkg = await readFile("./package.json", "utf8");
+const pkgnew = pkg.replace(/"version": ".+"/, `"version": "v${ver}"`);
+await writeFile("./package.json", pkgnew);
 
-const spago = await readFile('./spago.yaml', 'utf8')
-const spagonew = spago.replace(/version: .+/, `version: '${ver}'`)
-await writeFile('./spago.yaml', spagonew)
+const spago = await readFile("./spago.yaml", "utf8");
+const spagonew = spago.replace(/version: .+/, `version: '${ver}'`);
+await writeFile("./spago.yaml", spagonew);
 
-const readme = await readFile('./README.md', 'utf8')
+const readme = await readFile("./README.md", "utf8");
 const readmenew = readme.replace(
   /packages\/purescript-csv-stream\/.+?\//g,
   `/packages/purescript-csv-stream/${ver}/`,
-)
-await writeFile('./README.md', readmenew)
+);
+await writeFile("./README.md", readmenew);
 
-execSync(`git add spago.yaml package.json README.md`)
-execSync(`git commit -m 'chore: prepare v${ver}'`)
-execSync(`git tag v${ver}`)
-execSync(`git push --tags`)
-execSync(`git push --mirror github-mirror`)
+execSync(`git add spago.yaml package.json README.md`);
+execSync(`git commit -m 'chore: prepare v${ver}'`);
+execSync(`git tag v${ver}`);
+execSync(`git push --tags`);
+execSync(`git push --mirror github-mirror`);

package.json

@@ -1,6 +1,6 @@
 {
   "name": "purescript-csv-stream",
-  "version": "v1.1.1",
+  "version": "v1.2.3",
   "dependencies": {
     "csv-parse": "^5.5.5",
     "csv-stringify": "^6.4.6"

spago.lock

@@ -6,10 +6,12 @@ workspace:
         - aff: ">=7.1.0 <8.0.0"
         - arrays: ">=7.3.0 <8.0.0"
         - bifunctors: ">=6.0.0 <7.0.0"
+        - control: ">=6.0.0 <7.0.0"
         - datetime: ">=6.1.0 <7.0.0"
         - effect: ">=4.0.0 <5.0.0"
         - either: ">=6.1.0 <7.0.0"
         - exceptions: ">=6.0.0 <7.0.0"
+        - filterable: ">=5.0.0 <6.0.0"
         - foldable-traversable: ">=6.0.0 <7.0.0"
         - foreign: ">=7.0.0 <8.0.0"
         - foreign-object: ">=4.1.0 <5.0.0"
@@ -22,6 +24,8 @@ workspace:
         - node-streams: ">=9.0.0 <10.0.0"
         - nullable: ">=6.0.0 <7.0.0"
         - numbers: ">=9.0.1 <10.0.0"
+        - ordered-collections: ">=3.2.0 <4.0.0"
+        - parallel: ">=6.0.0 <7.0.0"
         - precise-datetime: ">=7.0.0 <8.0.0"
         - prelude: ">=6.0.1 <7.0.0"
         - record: ">=4.0.0 <5.0.0"
@@ -51,6 +55,7 @@ workspace:
         - enums
         - exceptions
         - exists
+        - filterable
         - fixed-points
         - foldable-traversable
         - foreign
@@ -268,6 +273,17 @@ packages:
     integrity: sha256-A0JQHpTfo1dNOj9U5/Fd3xndlRSE0g2IQWOGor2yXn8=
     dependencies:
       - unsafe-coerce
+  filterable:
+    type: registry
+    version: 5.0.0
+    integrity: sha256-cCojJHRnTmpY1j1kegI4CFwghdQ2Fm/8dzM8IlC+lng=
+    dependencies:
+      - arrays
+      - either
+      - foldable-traversable
+      - identity
+      - lists
+      - ordered-collections
   fixed-points:
     type: registry
     version: 7.0.0

spago.yaml

@@ -1,7 +1,7 @@
 package:
   name: csv-stream
   publish:
-    version: '1.1.1'
+    version: '1.2.3'
     license: 'GPL-3.0-or-later'
     location:
       githubOwner: 'cakekindel'
@@ -13,10 +13,12 @@ package:
     - aff: ">=7.1.0 <8.0.0"
     - arrays: ">=7.3.0 <8.0.0"
     - bifunctors: ">=6.0.0 <7.0.0"
+    - control: ">=6.0.0 <7.0.0"
     - datetime: ">=6.1.0 <7.0.0"
     - effect: ">=4.0.0 <5.0.0"
     - either: ">=6.1.0 <7.0.0"
     - exceptions: ">=6.0.0 <7.0.0"
+    - filterable: ">=5.0.0 <6.0.0"
     - foldable-traversable: ">=6.0.0 <7.0.0"
     - foreign: ">=7.0.0 <8.0.0"
     - foreign-object: ">=4.1.0 <5.0.0"
@@ -29,6 +31,8 @@ package:
     - node-streams: ">=9.0.0 <10.0.0"
     - nullable: ">=6.0.0 <7.0.0"
     - numbers: ">=9.0.1 <10.0.0"
+    - ordered-collections: ">=3.2.0 <4.0.0"
+    - parallel: ">=6.0.0 <7.0.0"
     - precise-datetime: ">=7.0.0 <8.0.0"
     - prelude: ">=6.0.1 <7.0.0"
     - record: ">=4.0.0 <5.0.0"

src/Data.CSV.Record.purs

@@ -7,8 +7,10 @@ import Control.Monad.Except (Except)
 import Data.Array as Array
 import Data.CSV (class ReadCSV, class WriteCSV, readCSV, writeCSV)
 import Data.List.NonEmpty (NonEmptyList)
+import Data.Map (Map)
+import Data.Map as Map
 import Data.Maybe (fromMaybe)
-import Data.Symbol (class IsSymbol)
+import Data.Symbol (class IsSymbol, reflectSymbol)
 import Foreign (ForeignError(..))
 import Prim.Row (class Cons, class Lacks)
 import Prim.RowList (class RowToList, Cons, Nil, RowList)
@@ -20,25 +22,29 @@ class RowToList r rl <= WriteCSVRecord r rl | rl -> r where
   writeCSVRecord :: { | r } -> Array String
 
 instance (RowToList r (Cons k v tailrl), IsSymbol k, WriteCSV v, Lacks k tail, Cons k v tail r, WriteCSVRecord tail tailrl) => WriteCSVRecord r (Cons k v tailrl) where
-  writeCSVRecord r = let
-    val = writeCSV $ Record.get (Proxy @k) r
-    tail = writeCSVRecord @tail @tailrl $ Record.delete (Proxy @k) r
-  in
-    [val] <> tail
+  writeCSVRecord r =
+    let
+      val = writeCSV $ Record.get (Proxy @k) r
+      tail = writeCSVRecord @tail @tailrl $ Record.delete (Proxy @k) r
+    in
+      [ val ] <> tail
 
 instance WriteCSVRecord () Nil where
   writeCSVRecord _ = []
 
 class ReadCSVRecord :: Row Type -> RowList Type -> Constraint
 class RowToList r rl <= ReadCSVRecord r rl | rl -> r where
-  readCSVRecord :: Array String -> Except (NonEmptyList ForeignError) { | r }
+  readCSVRecord :: Map String Int -> Array String -> Except (NonEmptyList ForeignError) { | r }
 
 instance (RowToList r (Cons k v tailrl), IsSymbol k, ReadCSV v, Lacks k tail, Cons k v tail r, ReadCSVRecord tail tailrl) => ReadCSVRecord r (Cons k v tailrl) where
-  readCSVRecord vals = do
-    valraw <- liftMaybe (pure $ ForeignError "unexpected end of record") $ Array.head vals
+  readCSVRecord cols vals = do
+    let
+      k = reflectSymbol (Proxy @k)
+    pos <- liftMaybe (pure $ ForeignError $ "row too long; did not expect value " <> k) $ Map.lookup k cols
+    let valraw = fromMaybe "" $ Array.index vals pos
     val <- readCSV @v valraw
-    tail <- readCSVRecord @tail @tailrl (fromMaybe [] $ Array.tail vals)
+    tail <- readCSVRecord @tail @tailrl cols vals
     pure $ Record.insert (Proxy @k) val tail
 
 instance ReadCSVRecord () Nil where
-  readCSVRecord _ = pure {}
+  readCSVRecord _ _ = pure {}

src/Data.CSV.purs

@@ -9,11 +9,12 @@ import Data.Int as Int
 import Data.List.NonEmpty (NonEmptyList)
 import Data.Maybe (Maybe(..), maybe)
 import Data.Newtype (unwrap)
+import Data.Number (fromString) as Number
 import Data.Number.Format (toString) as Number
 import Data.PreciseDateTime (fromDateTime, fromRFC3339String, toDateTimeLossy, toRFC3339String)
 import Data.RFC3339String (RFC3339String(..))
 import Data.String as String
-import Foreign (ForeignError(..), readInt, readNumber, unsafeToForeign)
+import Foreign (ForeignError(..))
 
 class ReadCSV a where
   readCSV :: String -> Except (NonEmptyList ForeignError) a
@@ -22,10 +23,10 @@ class WriteCSV a where
   writeCSV :: a -> String
 
 instance ReadCSV Int where
-  readCSV = readInt <<< unsafeToForeign
+  readCSV s = liftMaybe (pure $ ForeignError $ "invalid integer: " <> s) $ Int.fromString s
 
 instance ReadCSV Number where
-  readCSV = readNumber <<< unsafeToForeign
+  readCSV s = liftMaybe (pure $ ForeignError $ "invalid number: " <> s) $ Number.fromString s
 
 instance ReadCSV String where
   readCSV = pure

src/Node.Stream.CSV.Parse.js

@@ -1,7 +1,29 @@
-import {parse} from 'csv-parse'
+import { parse, Parser } from "csv-parse";
 
-/** @type {(s: import('csv-parse').Options) => () => import('csv-parse').Parser} */
-export const makeImpl = c => () => parse(c)
+class ParserWithColumns extends Parser {
+  /** @type {Array<string>} */
+  columns = [];
+  /** @type {Map<string, number> | null} */
+  columnsMap = null;
+}
 
-/** @type {(s: import('stream').Duplex) => () => string[] | null} */
-export const readImpl = s => () => s.read();
+/** @type {(s: import('csv-parse').Options) => () => ParserWithColumns} */
+export const makeImpl = (c) => () => {
+  const parser = new ParserWithColumns(c);
+  parser.once("readable", () => {
+    parser.columns = parser.read();
+  });
+  return parser;
+};
+
+/** @type {(s: ParserWithColumns) => () => Array<string> | null} */
+export const readImpl = (p) => () => p.read();
+
+/** @type {(s: ParserWithColumns) => () => Array<string>} */
+export const columnsArrayImpl = (p) => () => p.columns;
+
+/** @type {(s: ParserWithColumns) => () => Map<string, number> | null} */
+export const columnsMapImpl = (p) => () => p.columnsMap;
+
+/** @type {(s: ParserWithColumns) => (m: Map<string, number>) => () => void} */
+export const setColumnsMapImpl = (p) => (m) => () => (p.columnsMap = m);

src/Node.Stream.CSV.Parse.purs

@@ -1,22 +1,34 @@
 module Node.Stream.CSV.Parse where
 
-import Prelude
+import Prelude hiding (join)
 
+import Control.Alt ((<|>))
+import Control.Alternative (guard)
 import Control.Monad.Error.Class (liftEither)
 import Control.Monad.Except (runExcept)
 import Control.Monad.Maybe.Trans (MaybeT(..), runMaybeT)
-import Control.Monad.Rec.Class (whileJust)
+import Control.Monad.Rec.Class (class MonadRec, whileJust)
 import Control.Monad.ST.Global as ST
+import Control.Monad.Trans.Class (lift)
+import Control.MonadPlus (class Alternative)
+import Control.Parallel (class Parallel, parTraverse_)
+import Data.Array as Array
 import Data.Array.ST as Array.ST
 import Data.Bifunctor (lmap)
 import Data.CSV.Record (class ReadCSVRecord, readCSVRecord)
 import Data.Either (Either(..))
-import Data.Maybe (Maybe(..))
+import Data.Filterable (filter)
+import Data.Map (Map)
+import Data.Map as Map
+import Data.Maybe (Maybe(..), isNothing)
+import Data.Newtype (wrap)
 import Data.Nullable (Nullable)
 import Data.Nullable as Nullable
-import Data.Traversable (for_)
+import Data.Traversable (for)
 import Effect (Effect)
-import Effect.Aff (Aff, makeAff)
+import Effect as Effect
+import Effect.Aff (Canceler(..), delay, makeAff)
+import Effect.Aff.Class (class MonadAff, liftAff)
 import Effect.Class (liftEffect)
 import Effect.Exception (error)
 import Effect.Uncurried (mkEffectFn1)
@@ -77,10 +89,21 @@ type Config r =
 
 -- | Create a CSVParser
 make :: forall @r rl @config @missing @extra. RowToList r rl => ReadCSVRecord r rl => Union config missing (Config extra) => { | config } -> Effect (CSVParser r ())
-make = makeImpl <<< unsafeToForeign <<< Object.union (recordToForeign {columns: true, cast: false, cast_date: false}) <<< recordToForeign
+make = makeImpl <<< unsafeToForeign <<< Object.union (recordToForeign { columns: false, cast: false, cast_date: false }) <<< recordToForeign
 
 -- | Synchronously parse a CSV string
-parse :: forall @r rl @config missing extra. RowToList r rl => ReadCSVRecord r rl => Union config missing (Config extra) => { | config } -> String -> Aff (Array { | r })
+parse
+  :: forall @r rl @config missing extra m p
+   . Alternative p
+  => Parallel p m
+  => MonadAff m
+  => MonadRec m
+  => RowToList r rl
+  => ReadCSVRecord r rl
+  => Union config missing (Config extra)
+  => { | config }
+  -> String
+  -> m (Array { | r })
 parse config csv = do
   stream <- liftEffect $ make @r @config @missing @extra config
   void $ liftEffect $ Stream.writeString stream UTF8 csv
@@ -88,29 +111,68 @@ parse config csv = do
   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
+foreach
+  :: forall @r rl x m p
+   . Alternative p
+  => Parallel p m
+  => MonadRec m
+  => MonadAff m
+  => RowToList r rl
+  => ReadCSVRecord r rl
+  => CSVParser r x
+  -> ({ | r } -> m Unit)
+  -> m Unit
+foreach stream cb =
+  whileJust
+  $ runMaybeT
+  $ do
+      liftAff $ delay $ wrap 0.0
+
+      guard =<< not <$> liftEffect (Stream.closed stream)
+
+      isReadable <- liftEffect $ Stream.readable stream
+      liftAff $ when (not isReadable) $ makeAff \res -> do
+        stop <- flip (Event.once Stream.readableH) stream $ res $ Right unit
+        pure $ Canceler $ const $ liftEffect stop
+
+      recordsST <- liftEffect $ ST.toEffect $ Array.ST.new
+      liftEffect $ Effect.untilE do
+        r <- read @r stream
+        void $ for r $ ST.toEffect <<< flip Array.ST.push recordsST
+        pure $ isNothing r
+      records <- liftEffect $ ST.toEffect $ Array.ST.unsafeFreeze recordsST
+
+      lift $ parTraverse_ cb records
+      guard =<< not <$> liftEffect (Stream.closed stream)
+      pure unit
 
 -- | Reads a parsed record from the stream.
 -- |
 -- | Returns `Nothing` when either:
 -- | - The internal buffer of parsed records has been exhausted, but there will be more (`Node.Stream.readable` and `Node.Stream.closed` are both `false`)
 -- | - All records have been processed (`Node.Stream.closed` is `true`)
-read :: forall @r rl a. RowToList r rl => ReadCSVRecord r rl => CSVParser r a -> Effect (Maybe { | r })
+read
+  :: forall @r rl a
+   . RowToList r rl
+  => ReadCSVRecord r rl
+  => CSVParser r a
+  -> Effect (Maybe { | r })
 read stream = runMaybeT do
+  cols <- MaybeT $ getOrInitColumnsMap stream
   raw :: Array String <- MaybeT $ Nullable.toMaybe <$> readImpl stream
-  liftEither $ lmap (error <<< show) $ runExcept $ readCSVRecord @r @rl raw
+  liftEither $ lmap (error <<< show) $ runExcept $ readCSVRecord @r @rl cols raw
 
 -- | Collect all parsed records into an array
-readAll :: forall @r rl a. RowToList r rl => ReadCSVRecord r rl => CSVParser r a -> Aff (Array { | r })
+readAll
+  :: forall @r rl a m p
+   . Alternative p
+  => Parallel p m
+  => MonadRec m
+  => MonadAff m
+  => RowToList r rl
+  => ReadCSVRecord r rl
+  => CSVParser r a
+  -> m (Array { | r })
 readAll stream = do
   records <- liftEffect $ ST.toEffect $ Array.ST.new
   foreach stream $ void <<< liftEffect <<< ST.toEffect <<< flip Array.ST.push records
@@ -126,6 +188,30 @@ foreign import makeImpl :: forall r. Foreign -> Effect (Stream r)
 -- | FFI
 foreign import readImpl :: forall r. Stream r -> Effect (Nullable (Array String))
 
+-- | FFI
+foreign import columnsArrayImpl :: forall r. Stream r -> Effect (Array String)
+
+-- | FFI
+foreign import columnsMapImpl :: forall r. Stream r -> Effect (Nullable (Map String Int))
+
+-- | FFI
+foreign import setColumnsMapImpl :: forall r. Stream r -> Map String Int -> Effect Unit
+
+-- | FFI
+getOrInitColumnsMap :: forall r x. CSVParser r x -> Effect (Maybe (Map String Int))
+getOrInitColumnsMap s = runMaybeT do
+  cols :: Array String <- MaybeT $ filter (not <<< Array.null) <$> Just <$> columnsArrayImpl s
+  let
+    get = MaybeT $ Nullable.toMaybe <$> columnsMapImpl s
+    init = do
+      let
+        ixs = Array.range 0 (Array.length cols - 1)
+        assoc = Array.zip cols ixs
+        map = Map.fromFoldable assoc
+      lift $ setColumnsMapImpl s map
+      pure map
+  get <|> init
+
 -- | FFI
 recordToForeign :: forall r. Record r -> Object Foreign
 recordToForeign = unsafeCoerce

src/Node.Stream.CSV.Stringify.js

@@ -1,7 +1,7 @@
-import {stringify} from 'csv-stringify'
+import { stringify } from "csv-stringify";
 
 /** @type {(c: import('csv-stringify').Options) => () => import('csv-stringify').Stringifier} */
-export const makeImpl = c => () => stringify(c)
+export const makeImpl = (c) => () => stringify(c);
 
 /** @type {(s: import('csv-stringify').Stringifier) => (vals: Array<string>) => () => void} */
-export const writeImpl = s => vals => () => s.write(vals)
+export const writeImpl = (s) => (vals) => () => s.write(vals);

src/Node.Stream.CSV.Stringify.purs

@@ -2,7 +2,7 @@ module Node.Stream.CSV.Stringify where
 
 import Prelude
 
-import Control.Monad.Rec.Class (whileJust)
+import Control.Monad.Rec.Class (class MonadRec, whileJust)
 import Control.Monad.ST.Global as ST
 import Data.Array as Array
 import Data.Array.ST as Array.ST
@@ -13,7 +13,8 @@ import Data.Maybe (Maybe(..))
 import Data.String.Regex (Regex)
 import Data.Traversable (for_)
 import Effect (Effect)
-import Effect.Aff (Aff, makeAff)
+import Effect.Aff (Canceler(..), makeAff)
+import Effect.Aff.Class (class MonadAff, liftAff)
 import Effect.Class (liftEffect)
 import Foreign (Foreign, unsafeToForeign)
 import Foreign.Object (Object)
@@ -64,10 +65,10 @@ recordToForeign = unsafeCoerce
 
 -- | Create a CSVStringifier
 make :: forall @r rl @config @missing @extra. Keys rl => RowToList r rl => WriteCSVRecord r rl => Union config missing (Config extra) => { | config } -> Effect (CSVStringifier r ())
-make = makeImpl <<< unsafeToForeign <<< Object.union (recordToForeign {columns: Array.fromFoldable $ keys (Proxy @r)}) <<< recordToForeign
+make = makeImpl <<< unsafeToForeign <<< Object.union (recordToForeign { columns: Array.fromFoldable $ keys (Proxy @r) }) <<< recordToForeign
 
 -- | Synchronously stringify a collection of records
-stringify :: forall @r rl f @config missing extra. Keys rl => Foldable f => RowToList r rl => WriteCSVRecord r rl => Union config missing (Config extra) => { | config } -> f { | r } -> Aff String
+stringify :: forall @r rl f m @config missing extra. MonadAff m => MonadRec m => Keys rl => Foldable f => RowToList r rl => WriteCSVRecord r rl => Union config missing (Config extra) => { | config } -> f { | r } -> m String
 stringify config records = do
   stream <- liftEffect $ make @r @config @missing @extra config
   liftEffect $ for_ records \r -> write stream r
@@ -82,10 +83,12 @@ write :: forall @r rl a. RowToList r rl => WriteCSVRecord r rl => CSVStringifier
 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 :: forall m r x. MonadAff m => MonadRec m => CSVStringifier r x -> (String -> m Unit) -> m 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
+  liftAff $ when (not isReadable) $ makeAff \res -> do
+    stop <- flip (Event.once Stream.readableH) stream $ res $ Right unit
+    pure $ Canceler $ const $ liftEffect stop
   whileJust do
     s <- liftEffect $ (join <<< map blush) <$> Stream.readEither stream
     for_ s cb
@@ -94,7 +97,7 @@ foreach stream cb = whileJust do
   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 :: forall r a m. MonadAff m => MonadRec m => CSVStringifier r a -> m String
 readAll stream = do
   chunks <- liftEffect $ ST.toEffect $ Array.ST.new
   foreach stream $ void <<< liftEffect <<< ST.toEffect <<< flip Array.ST.push chunks