chore: prepare v1.3.3

package.json

@@ -1,6 +1,6 @@
 {
   "name": "purescript-node-stream-pipes",
-  "version": "v1.3.0",
+  "version": "v1.3.3",
   "type": "module",
   "dependencies": {
     "csv-parse": "^5.5.5",

spago.yaml

@@ -1,7 +1,7 @@
 package:
   name: node-stream-pipes
   publish:
-    version: '1.3.0'
+    version: '1.3.3'
     license: 'GPL-3.0-or-later'
     location:
       githubOwner: 'cakekindel'

src/Pipes.Node.FS.purs

@@ -8,7 +8,7 @@ import Effect.Aff.Class (class MonadAff)
 import Effect.Class (liftEffect)
 import Effect.Exception (Error)
 import Node.Buffer (Buffer)
-import Node.FS.Stream (WriteStreamOptions)
+import Node.FS.Stream (WriteStreamOptions, ReadStreamOptions)
 import Node.FS.Stream as FS.Stream
 import Node.Path (FilePath)
 import Node.Stream.Object as O
@@ -61,3 +61,19 @@ read :: forall m. MonadAff m => MonadThrow Error m => FilePath -> Producer (Mayb
 read p = do
   r <- liftEffect $ FS.Stream.createReadStream p
   fromReadable $ O.fromBufferReadable r
+
+-- | Creates a `fs.Readable` stream for the file at the given path.
+-- |
+-- | Emits `Nothing` before closing. To opt out of this behavior,
+-- | use `Pipes.Node.Stream.withoutEOS` or `Pipes.Node.Stream.unEOS`.
+read'
+  :: forall r trash m
+   . Union r trash ReadStreamOptions
+  => MonadAff m
+  => MonadThrow Error m
+  => Record r
+  -> FilePath
+  -> Producer (Maybe Buffer) m Unit
+read' opts p = do
+  r <- liftEffect $ FS.Stream.createReadStream' p opts
+  fromReadable $ O.fromBufferReadable r

src/Pipes.Node.Stream.purs

@@ -94,17 +94,21 @@ fromTransform t =
       liftEffect $ removeErrorListener
       fromReadable t
       pure $ Done unit
-    yieldFromReadableHalf = do
-      res <- liftEffect (O.read t)
-      case res of
-        O.ReadJust a -> yield (Just a)
-        O.ReadWouldBlock -> pure unit
-        O.ReadClosed -> yield Nothing *> pure unit
+    yieldFromReadableHalf =
+      flip tailRecM unit $ const do
+        res <- liftEffect (O.read t)
+        case res of
+          O.ReadJust a -> do
+            yield $ Just a
+            pure $ Loop unit
+          O.ReadWouldBlock -> pure $ Done unit
+          O.ReadClosed -> yield Nothing $> Done unit
     go { error, cancel } = do
       liftAff $ delay $ wrap 0.0
       err <- liftEffect $ liftST $ STRef.read error
       for_ err throwError
 
+      yieldFromReadableHalf
       ma <- await
       case ma of
         Nothing -> cleanup cancel