chore: prepare v1.2.10

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.11",
+  "version": "v1.2.10",
   "dependencies": {
     "csv-parse": "^5.5.5",
     "csv-stringify": "^6.4.6"

spago.lock

@@ -34,6 +34,7 @@ workspace:
         - tailrec: ">=6.1.0 <7.0.0"
         - transformers: ">=6.0.0 <7.0.0"
         - typelevel-prelude: ">=7.0.0 <8.0.0"
+        - unlift: ">=1.0.1 <2.0.0"
         - unsafe-coerce: ">=6.0.0 <7.0.0"
       test_dependencies:
         - console
@@ -42,6 +43,7 @@ workspace:
         - arraybuffer-types
         - arrays
         - bifunctors
+        - catenable-lists
         - console
         - const
         - contravariant
@@ -60,6 +62,8 @@ workspace:
         - foreign
         - foreign-object
         - formatters
+        - free
+        - freet
         - functions
         - functors
         - gen
@@ -70,6 +74,7 @@ workspace:
         - lazy
         - lists
         - maybe
+        - monad-control
         - newtype
         - node-buffer
         - node-event-emitter
@@ -99,6 +104,7 @@ workspace:
         - typelevel-prelude
         - unfoldable
         - unicode
+        - unlift
         - unsafe-coerce
   extra_packages: {}
 packages:
@@ -158,6 +164,18 @@ packages:
       - newtype
       - prelude
       - tuples
+  catenable-lists:
+    type: registry
+    version: 7.0.0
+    integrity: sha256-76vYENhwF4BWTBsjeLuErCH2jqVT4M3R1HX+4RwSftA=
+    dependencies:
+      - control
+      - foldable-traversable
+      - lists
+      - maybe
+      - prelude
+      - tuples
+      - unfoldable
   console:
     type: registry
     version: 6.1.0
@@ -351,6 +369,40 @@ packages:
       - parsing
       - prelude
       - transformers
+  free:
+    type: registry
+    version: 7.1.0
+    integrity: sha256-JAumgEsGSzJCNLD8AaFvuX7CpqS5yruCngi6yI7+V5k=
+    dependencies:
+      - catenable-lists
+      - control
+      - distributive
+      - either
+      - exists
+      - foldable-traversable
+      - invariant
+      - lazy
+      - maybe
+      - prelude
+      - tailrec
+      - transformers
+      - tuples
+      - unsafe-coerce
+  freet:
+    type: registry
+    version: 7.0.0
+    integrity: sha256-zkL6wU4ZPq8xz1kGFxoliWqyhBksepMJTyA68VEBaJo=
+    dependencies:
+      - aff
+      - bifunctors
+      - effect
+      - either
+      - exists
+      - free
+      - prelude
+      - tailrec
+      - transformers
+      - tuples
   functions:
     type: registry
     version: 6.0.0
@@ -460,6 +512,15 @@ packages:
       - invariant
       - newtype
       - prelude
+  monad-control:
+    type: registry
+    version: 5.0.0
+    integrity: sha256-bgfDW30wbIm70NR1Tvvh9P+VFQMDh1wK2sSJXCj/dZc=
+    dependencies:
+      - aff
+      - freet
+      - identity
+      - lists
   newtype:
     type: registry
     version: 5.0.0
@@ -783,6 +844,23 @@ packages:
       - foldable-traversable
       - maybe
       - strings
+  unlift:
+    type: registry
+    version: 1.0.1
+    integrity: sha256-nbBCVV0fZz/3UHKoW11dcTwBYmQOIgK31ht2BN47RPw=
+    dependencies:
+      - aff
+      - effect
+      - either
+      - freet
+      - identity
+      - lists
+      - maybe
+      - monad-control
+      - prelude
+      - st
+      - transformers
+      - tuples
   unsafe-coerce:
     type: registry
     version: 6.0.0

spago.yaml

@@ -1,7 +1,7 @@
 package:
   name: csv-stream
   publish:
-    version: '1.1.11'
+    version: '1.2.10'
     license: 'GPL-3.0-or-later'
     location:
       githubOwner: 'cakekindel'
@@ -41,6 +41,7 @@ package:
     - tailrec: ">=6.1.0 <7.0.0"
     - transformers: ">=6.0.0 <7.0.0"
     - typelevel-prelude: ">=7.0.0 <8.0.0"
+    - unlift: ">=1.0.1 <2.0.0"
     - unsafe-coerce: ">=6.0.0 <7.0.0"
   test:
     main: Test.Main

src/Data.CSV.Record.purs

@@ -22,11 +22,12 @@ 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 _ = []

src/Node.Stream.CSV.Parse.js

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

src/Node.Stream.CSV.Parse.purs

@@ -1,13 +1,13 @@
 module Node.Stream.CSV.Parse where
 
-import Prelude
+import Prelude hiding (join)
 
 import Control.Alt ((<|>))
-import Control.Monad.Error.Class (liftEither)
+import Control.Monad.Error.Class (liftEither, liftMaybe)
 import Control.Monad.Except (runExcept)
+import Control.Monad.Except.Trans (catchError)
 import Control.Monad.Maybe.Trans (MaybeT(..), runMaybeT)
-import Control.Monad.Rec.Class (class MonadRec, whileJust)
-import Control.Monad.ST.Global as ST
+import Control.Monad.ST.Class (liftST)
 import Control.Monad.Trans.Class (lift)
 import Data.Array as Array
 import Data.Array.ST as Array.ST
@@ -18,12 +18,13 @@ import Data.Filterable (filter)
 import Data.Map (Map)
 import Data.Map as Map
 import Data.Maybe (Maybe(..))
+import Data.Newtype (wrap)
 import Data.Nullable (Nullable)
 import Data.Nullable as Nullable
-import Data.Traversable (for_)
 import Effect (Effect)
-import Effect.Aff (Canceler(..), makeAff)
+import Effect.Aff (Canceler(..), delay, launchAff_, makeAff)
 import Effect.Aff.Class (class MonadAff, liftAff)
+import Effect.Aff.Unlift (class MonadUnliftAff, UnliftAff(..), askUnliftAff)
 import Effect.Class (liftEffect)
 import Effect.Exception (error)
 import Effect.Uncurried (mkEffectFn1)
@@ -84,10 +85,19 @@ 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: false, 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 m. MonadAff m => MonadRec m => RowToList r rl => ReadCSVRecord r rl => Union config missing (Config extra) => { | config } -> String -> m (Array { | r })
+parse
+  :: forall @r rl @config missing extra m
+   . MonadUnliftAff m
+  => MonadAff 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
@@ -95,41 +105,73 @@ 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 m. MonadRec m => MonadAff m => RowToList r rl => ReadCSVRecord r rl => CSVParser r x -> ({ | r } -> m Unit) -> m Unit
-foreach stream cb = whileJust do
-  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
-  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
+   . MonadUnliftAff m
+  => MonadAff m
+  => RowToList r rl
+  => ReadCSVRecord r rl
+  => CSVParser r x
+  -> ({ | r } -> m Unit)
+  -> m Unit
+foreach stream cb = do
+  UnliftAff unlift <- askUnliftAff
+
+  liftAff $ makeAff \res -> pure mempty <* flip (Event.once columnsH) stream $ const do
+    void $ getOrInitColumnsMap stream
+    res $ Right unit
+
+  liftAff $ makeAff \res -> do
+    removeDataListener <- flip (Event.on dataH) stream \row -> launchAff_ $ delay (wrap 0.0) <* liftEffect do
+      cols <- liftMaybe (error "unreachable") =<< getOrInitColumnsMap stream
+      record <- liftEither $ lmap (error <<< show) $ runExcept $ readCSVRecord @r @rl cols row
+      launchAff_ $ flip catchError (liftEffect <<< res <<< Left) (unlift $ cb record)
+    removeEndListener <- flip (Event.once Stream.endH) stream (res $ Right unit)
+    removeErrorListener <- flip (Event.on Stream.errorH) stream (res <<< Left)
+
+    pure $ Canceler $ const $ liftEffect do
+      removeDataListener
+      removeEndListener
+      removeErrorListener
 
 -- | 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 cols raw
 
 -- | Collect all parsed records into an array
-readAll :: forall @r rl a m. MonadRec m => MonadAff m => RowToList r rl => ReadCSVRecord r rl => CSVParser r a -> m (Array { | r })
+readAll
+  :: forall @r rl a m
+   . MonadUnliftAff 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
-  liftEffect $ ST.toEffect $ Array.ST.unsafeFreeze records
+  records <- liftEffect $ liftST $ Array.ST.new
+  foreach stream $ void <<< liftEffect <<< liftST <<< flip Array.ST.push records
+  liftEffect $ liftST $ Array.ST.unsafeFreeze records
 
 -- | `data` event. Emitted when a CSV record has been parsed.
-dataH :: forall r a. EventHandle1 (CSVParser r a) { | r }
+dataH :: forall r a. EventHandle1 (CSVParser r a) (Array String)
 dataH = EventHandle "data" mkEffectFn1
 
+-- | `columns` event. Emitted when the header row has been parsed.
+columnsH :: forall r a. EventHandle1 (CSVParser r a) (Array String)
+columnsH = EventHandle "columns" mkEffectFn1
+
 -- | FFI
 foreign import makeImpl :: forall r. Foreign -> Effect (Stream r)
 

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

@@ -65,7 +65,7 @@ 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 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