From 52dd297c9ccbbb1223ac921e0c1944c45e36f36d Mon Sep 17 00:00:00 2001
From: Orion Kindel <orionkindel@gmail.com>
Date: Fri, 3 May 2024 14:22:07 -0500
Subject: [PATCH] fix: what

---
 src/Node.Stream.CSV.Parse.purs | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/src/Node.Stream.CSV.Parse.purs b/src/Node.Stream.CSV.Parse.purs
index f568b2a..ecb3d9d 100644
--- a/src/Node.Stream.CSV.Parse.purs
+++ b/src/Node.Stream.CSV.Parse.purs
@@ -120,9 +120,11 @@ foreach stream cb = do
   alreadyHaveCols <- liftEffect $ getOrInitColumnsMap stream
   when (isNothing alreadyHaveCols)
     $ liftAff
-    $ makeAff \res -> pure mempty <* flip (Event.once columnsH) stream $ const do
-    void $ getOrInitColumnsMap stream
-    res $ Right unit
+    $ 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 -> launchAff_ $ delay (wrap 0.0) <* liftEffect do