chore: prepare v1.2.16

package.json

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

spago.lock

@@ -16,6 +16,7 @@ workspace:
         - foreign: ">=7.0.0 <8.0.0"
         - foreign-object: ">=4.1.0 <5.0.0"
         - integers: ">=6.0.0 <7.0.0"
+        - js-timers
         - lists: ">=7.0.0 <8.0.0"
         - maybe: ">=6.0.0 <7.0.0"
         - newtype: ">=5.0.0 <6.0.0"
@@ -71,6 +72,7 @@ workspace:
         - integers
         - invariant
         - js-date
+        - js-timers
         - lazy
         - lists
         - maybe
@@ -477,6 +479,12 @@ packages:
       - foreign
       - integers
       - now
+  js-timers:
+    type: registry
+    version: 6.1.0
+    integrity: sha256-znHWLSSOYw15P5DTcsAdal2lf7nGA2yayLdOZ2t5r7o=
+    dependencies:
+      - effect
   lazy:
     type: registry
     version: 6.0.0

spago.yaml

@@ -1,7 +1,7 @@
 package:
   name: csv-stream
   publish:
-    version: '1.2.8'
+    version: '1.2.16'
     license: 'GPL-3.0-or-later'
     location:
       githubOwner: 'cakekindel'
@@ -10,6 +10,7 @@ package:
     strict: true
     pedanticPackages: true
   dependencies:
+    - js-timers
     - aff: ">=7.1.0 <8.0.0"
     - arrays: ">=7.3.0 <8.0.0"
     - bifunctors: ">=6.0.0 <7.0.0"

src/Node.Stream.CSV.Parse.js

@@ -12,6 +12,7 @@ export const makeImpl = (c) => () => {
   const parser = new ParserWithColumns(c);
   parser.once("readable", () => {
     parser.columns = parser.read();
+    parser.emit("columns", parser.columns);
   });
   return parser;
 };

src/Node.Stream.CSV.Parse.purs

@@ -3,7 +3,7 @@ module Node.Stream.CSV.Parse where
 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)
@@ -15,18 +15,17 @@ import Data.Bifunctor (lmap)
 import Data.CSV.Record (class ReadCSVRecord, readCSVRecord)
 import Data.Either (Either(..))
 import Data.Filterable (filter)
-import Data.Foldable (for_)
 import Data.Map (Map)
 import Data.Map as Map
-import Data.Maybe (Maybe(..))
+import Data.Maybe (Maybe(..), isNothing)
 import Data.Nullable (Nullable)
 import Data.Nullable as Nullable
 import Effect (Effect)
-import Effect.Aff (Canceler(..), launchAff_, makeAff)
+import Effect.Aff (Canceler(..), 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.Timer (setTimeout)
 import Effect.Uncurried (mkEffectFn1)
 import Foreign (Foreign, unsafeToForeign)
 import Foreign.Object (Object)
@@ -90,8 +89,7 @@ make = makeImpl <<< unsafeToForeign <<< Object.union (recordToForeign { columns:
 -- | Synchronously parse a CSV string
 parse
   :: forall @r rl @config missing extra m
-   . MonadUnliftAff m
-  => MonadAff m
+   . MonadAff m
   => RowToList r rl
   => ReadCSVRecord r rl
   => Union config missing (Config extra)
@@ -107,23 +105,34 @@ parse config csv = do
 -- | Loop until the stream is closed, invoking the callback with each record as it is parsed.
 foreach
   :: forall @r rl x m
-   . MonadUnliftAff m
-  => MonadAff m
+   . MonadAff m
   => RowToList r rl
   => ReadCSVRecord r rl
   => CSVParser r x
-  -> ({ | r } -> m Unit)
+  -> ({ | r } -> Effect Unit)
   -> m Unit
 foreach stream cb = do
-  UnliftAff unlift <- askUnliftAff
+  alreadyHaveCols <- liftEffect $ getOrInitColumnsMap stream
+  when (isNothing alreadyHaveCols)
+    $ liftAff
+    $ makeAff \res -> do
+        stop <- flip (Event.once columnsH) stream $ const do
+          void $ getOrInitColumnsMap stream
+          res $ Right unit
+        pure $ Canceler $ const $ liftEffect stop
+
   liftAff $ makeAff \res -> do
-    removeDataListener <- flip (Event.on dataH) stream \row -> do
-      cols <- getOrInitColumnsMap stream
-      for_ cols \cols' -> do
-        record <- liftEither $ lmap (error <<< show) $ runExcept $ readCSVRecord @r @rl cols' row
-        launchAff_ $ flip catchError (liftEffect <<< res <<< Left) (unlift $ cb record)
+    removeDataListener <- flip (Event.on dataH) stream \row ->
+      void
+        $ setTimeout 0
+        $ flip catchError (res <<< Left)
+        $ do
+            cols <- liftMaybe (error "unreachable") =<< getOrInitColumnsMap stream
+            record <- liftEither $ lmap (error <<< show) $ runExcept $ readCSVRecord @r @rl cols row
+            flip catchError (liftEffect <<< res <<< Left) (cb record)
+
     removeEndListener <- flip (Event.once Stream.endH) stream (res $ Right unit)
-    removeErrorListener <- flip (Event.on Stream.errorH) stream (res <<< Left)
+    removeErrorListener <- flip (Event.once Stream.errorH) stream (res <<< Left)
 
     pure $ Canceler $ const $ liftEffect do
       removeDataListener
@@ -149,8 +158,7 @@ read stream = runMaybeT do
 -- | Collect all parsed records into an array
 readAll
   :: forall @r rl a m
-   . MonadUnliftAff m
-  => MonadAff m
+   . MonadAff m
   => RowToList r rl
   => ReadCSVRecord r rl
   => CSVParser r a
@@ -164,6 +172,10 @@ readAll stream = do
 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)