wip: explore removing delays(8)

src/Pipes.Node.Stream.purs

@@ -13,6 +13,7 @@ import Data.Tuple.Nested ((/\))
 import Effect.Aff.Class (class MonadAff, liftAff)
 import Effect.Class (liftEffect)
 import Effect.Exception (Error)
+import Node.Stream.Object (needsDrain)
 import Node.Stream.Object as O
 import Pipes (await, yield)
 import Pipes (for) as P
@@ -108,23 +109,25 @@ fromTransform t =
       err <- liftEffect $ liftST $ STRef.read error
       for_ err throwError
 
-      yieldWhileReadable
+      needsDrain <- liftEffect $ O.needsDrain t
-      ma <- await
+      if needsDrain then do
-      case ma of
+        liftAff $ O.awaitReadableOrClosed t
-        Nothing -> cleanup cancel
+        yieldWhileReadable
-        Just a' -> do
+        pure $ Loop {error, cancel}
-          res <- liftEffect $ O.write t a'
+      else do
-          case res of
+        ma <- await
-            O.WriteClosed -> cleanup cancel
+        case ma of
-            O.WriteOk -> do
+          Nothing -> cleanup cancel
-              maybeYield1
+          Just a' -> do
-              pure $ Loop { error, cancel }
+            res <- liftEffect $ O.write t a'
-            O.WriteWouldBlock -> do
+            case res of
-              queueLen <- liftEffect $ O.readableLength t
+              O.WriteClosed -> cleanup cancel
-              when (queueLen == 0) $ liftAff $ O.awaitReadableOrClosed t
+              O.WriteOk -> do
-              yieldWhileReadable
+                maybeYield1
-              liftAff $ O.awaitWritableOrClosed t
+                pure $ Loop { error, cancel }
-              pure $ Loop { error, cancel }
+              O.WriteWouldBlock -> do
+                yieldWhileReadable
+                pure $ Loop { error, cancel }
   in
     do
       r <- liftEffect $ O.withErrorST t