Merge pull request #6 from jamesdbrock/improve

Read from zero-length Readable.

Author: jamesdbrock

CHANGELOG.md

@@ -12,6 +12,12 @@ Bugfixes:
 
 Other improvements:
 
+## [Unreleased]
+
+Bugfixes:
+
+* Read from zero-length `Readable`.
+
 # v3.0.0
 
 Breaking changes:

spago.dhall

@@ -17,7 +17,9 @@ to generate this file without the comments in this block.
     , "exceptions"
     , "node-buffer"
     , "node-streams"
+    , "nullable"
     , "st"
+    , "refs"
     , "arrays"
     , "either"
     , "maybe"

src/Node/Stream/Aff.purs

@@ -87,7 +87,6 @@ module Node.Stream.Aff
 import Prelude
 
 import Control.Monad.ST.Class (liftST)
-import Control.Monad.ST.Ref as ST.Ref
 import Data.Array as Array
 import Data.Array.ST as Array.ST
 import Data.Either (Either(..))
@@ -98,6 +97,7 @@ import Effect.Aff (effectCanceler, makeAff, nonCanceler)
 import Effect.Aff.Class (class MonadAff, liftAff)
 import Effect.Class (class MonadEffect, liftEffect)
 import Effect.Exception (catchException)
+import Effect.Ref as Ref
 import Node.Buffer (Buffer)
 import Node.Buffer as Buffer
 import Node.Encoding as Encoding
@@ -110,7 +110,6 @@ import Node.Stream.Aff.Internal (onceDrain, onceEnd, onceError, onceReadable, re
 -- |
 -- | This function is useful for streams like __stdin__ which never
 -- | reach End-Of-File.
--- |
 readSome
   :: forall m r
    . MonadAff m
@@ -121,6 +120,11 @@ readSome r = liftAff <<< makeAff $ \res -> do
 
   removeError <- onceError r $ res <<< Left
 
+  removeEnd <- onceEnd r do
+    removeError
+    ret <- liftST $ Array.ST.unsafeFreeze bufs
+    res (Right (Tuple ret false))
+
   -- try to read right away.
   catchException (res <<< Left) do
     ifM (readable r)
@@ -133,15 +137,19 @@ readSome r = liftAff <<< makeAff $ \res -> do
               pure false
       do
         removeError
+        removeEnd
         res (Right (Tuple [] false))
 
 
   ret1 <- liftST $ Array.ST.unsafeFreeze bufs
-  removeReadable <- if Array.length ret1 == 0 then do
-    -- if we couldn't read anything right away, then wait until the stream is readable.
+  readagain <- readable r
+  removeReadable <- if readagain && Array.length ret1 == 0 then do
+    -- if still readable and we couldn't read anything right away,
+    -- then wait for the readable event.
     -- “The 'readable' event will also be emitted once the end of the
     -- stream data has been reached but before the 'end' event is emitted.”
-    -- We already checked the `readable` property so we don't have to check again.
+    -- if not readable then this was a zero-length Readable stream.
+    -- https://nodejs.org/api/stream.html#event-readable
     onceReadable r do
       catchException (res <<< Left) do
         untilE do
@@ -152,23 +160,24 @@ readSome r = liftAff <<< makeAff $ \res -> do
               pure false
         ret2 <- liftST $ Array.ST.unsafeFreeze bufs
         removeError
-        readagain <- readable r
-        res (Right (Tuple ret2 readagain))
+        removeEnd
+        readagain2 <- readable r
+        res (Right (Tuple ret2 readagain2))
 
     -- return what we read right away
   else do
     removeError
-    readagain <- readable r
+    removeEnd
     res (Right (Tuple ret1 readagain))
     pure (pure unit) -- dummy canceller
 
   -- canceller might by called while waiting for `onceReadable`
   pure $ effectCanceler do
     removeError
+    removeEnd
     removeReadable
 
 
-
 -- | Read all data until the end of the stream.
 -- |
 -- | Note that __stdin__ will never end.
@@ -179,7 +188,7 @@ readAll
   -> m (Tuple (Array Buffer) Boolean)
 readAll r = liftAff <<< makeAff $ \res -> do
   bufs <- liftST $ Array.ST.new
-  removeReadable <- liftST $ ST.Ref.new (pure unit :: Effect Unit)
+  removeReadable <- Ref.new (pure unit :: Effect Unit)
 
   removeError <- onceError r $ res <<< Left
 
@@ -192,7 +201,7 @@ readAll r = liftAff <<< makeAff $ \res -> do
     cleanupRethrow err = do
       removeError
       removeEnd
-      join $ liftST $ ST.Ref.read removeReadable
+      join $ Ref.read removeReadable
       res (Left err)
 
   -- try to read right away.
@@ -224,15 +233,15 @@ readAll r = liftAff <<< makeAff $ \res -> do
                 _ <- liftST $ Array.ST.push chunk bufs
                 pure false
           waitToRead -- this is not recursion
-      void $ liftST $ ST.Ref.write removeReadable' removeReadable
+      Ref.write removeReadable' removeReadable
 
   waitToRead
 
   -- canceller might by called while waiting for `onceReadable`
   pure $ effectCanceler do
     removeError
     removeEnd
-    join $ liftST $ ST.Ref.read removeReadable
+    join $ Ref.read removeReadable
 
 
 -- | Wait for *N* bytes to become available from the stream.
@@ -249,9 +258,9 @@ readN
   -> Int
   -> m (Tuple (Array Buffer) Boolean)
 readN r n = liftAff <<< makeAff $ \res -> do
-  redRef <- liftST $ ST.Ref.new 0
+  redRef <- Ref.new 0
   bufs <- liftST $ Array.ST.new
-  removeReadable <- liftST $ ST.Ref.new (pure unit :: Effect Unit)
+  removeReadable <- Ref.new (pure unit :: Effect Unit)
 
   -- TODO on error, we're not calling removeEnd...
   removeError <- onceError r $ res <<< Left
@@ -260,22 +269,21 @@ readN r n = liftAff <<< makeAff $ \res -> do
   -- if there are more bytes in the stream?
   removeEnd <- onceEnd r do
     removeError
-    -- join $ liftST $ ST.Ref.read removeReadable
     ret <- liftST $ Array.ST.unsafeFreeze bufs
     res (Right (Tuple ret false))
 
   let
     cleanupRethrow err = do
       removeError
       removeEnd
-      join $ liftST $ ST.Ref.read removeReadable
+      join $ Ref.read removeReadable
       res (Left err)
 
     -- try to read N bytes and then either return N bytes or run a continuation
     tryToRead continuation = do
       catchException cleanupRethrow do
         untilE do
-          red <- liftST $ ST.Ref.read redRef
+          red <- Ref.read redRef
           -- https://nodejs.org/docs/latest-v15.x/api/stream.html#stream_readable_read_size
           -- “If size bytes are not available to be read, null will be returned
           -- unless the stream has ended, in which case all of the data remaining
@@ -285,12 +293,12 @@ readN r n = liftAff <<< makeAff $ \res -> do
             Just chunk -> do
               _ <- liftST $ Array.ST.push chunk bufs
               s <- Buffer.size chunk
-              red' <- liftST $ ST.Ref.modify (_+s) redRef
+              red' <- Ref.modify (_+s) redRef
               if red' >= n then
                 pure true
               else
                 pure false
-        red <- liftST $ ST.Ref.read redRef
+        red <- Ref.read redRef
         if red >= n then do
           removeError
           removeEnd
@@ -314,14 +322,14 @@ readN r n = liftAff <<< makeAff $ \res -> do
     waitToRead _ = do
       removeReadable' <- onceReadable r do
         tryToRead waitToRead
-      void $ liftST $ ST.Ref.write removeReadable' removeReadable
+      Ref.write removeReadable' removeReadable
   waitToRead unit
 
   -- canceller might by called while waiting for `onceReadable`
   pure $ effectCanceler do
     removeError
     removeEnd
-    join $ liftST $ ST.Ref.read removeReadable
+    join $ Ref.read removeReadable
 
 
 -- | Write to a stream.
@@ -335,7 +343,7 @@ write
   -> m Unit
 write w bs = liftAff <<< makeAff $ \res -> do
   bufs <- liftST $ Array.ST.thaw bs
-  removeDrain <- liftST $ ST.Ref.new (pure unit :: Effect Unit)
+  removeDrain <- Ref.new (pure unit :: Effect Unit)
 
   removeError <- onceError w $ res <<< Left
 
@@ -366,15 +374,15 @@ write w bs = liftAff <<< makeAff $ \res -> do
                 pure false
               else do
                 removeDrain' <- onceDrain w oneWrite
-                void $ liftST $ ST.Ref.write removeDrain' removeDrain
+                void $ Ref.write removeDrain' removeDrain
                 pure true
 
   oneWrite
 
   -- canceller might be called while waiting for `onceDrain`
   pure $ effectCanceler do
     removeError
-    join $ liftST $ ST.Ref.read removeDrain
+    join $ Ref.read removeDrain
 
 -- | Signal that no more data will be written to the `Writable`. Will complete
 -- | after all data is written and flushed.

src/Node/Stream/Internal.js

@@ -1,3 +1,5 @@
+import stream from 'stream';
+
 export const onceReadable = s => f => () => {
   s.once('readable', f);
   return () => {s.removeListener('readable', f);};
@@ -19,5 +21,13 @@ export const onceError = s => f => () => {
 }
 
 export const readable = s => () => {
-    return s.readable;
+  return s.readable;
+}
+
+export const push = s => buf => () => {
+  return s.push(buf);
 }
+
+export const newReadable = () => {
+  return new stream.Readable();
+}

src/Node/Stream/Internal.purs

@@ -7,13 +7,20 @@ module Node.Stream.Aff.Internal
   , onceError
   , onceReadable
   , readable
+  , push
+  , newReadable
+  , newReadableStringUTF8
   )
   where
 
 import Prelude
 
+import Data.Nullable (Nullable, notNull, null)
 import Effect (Effect)
 import Effect.Exception (Error)
+import Node.Buffer (Buffer)
+import Node.Buffer as Buffer
+import Node.Encoding as Encoding
 import Node.Stream (Readable, Stream, Writable)
 
 -- | Listen for one `readable` event, call the callback, then remove
@@ -69,3 +76,27 @@ foreign import readable
   :: forall r
    . Readable r
   -> Effect Boolean
+
+
+-- | [`readable.push(chunk[, encoding])`](https://nodejs.org/api/stream.html#readablepushchunk-encoding)
+foreign import push
+  :: forall r
+   . Readable r
+  -> Nullable Buffer
+  -> Effect Boolean
+
+-- | `new stream.Readable()`
+foreign import newReadable
+  :: forall r
+   . Effect (Readable r)
+
+-- | Construct a `Readable` from a `String`.
+newReadableStringUTF8
+  :: forall r
+   . String
+  -> Effect (Readable r)
+newReadableStringUTF8 strng = do
+  rstream <- newReadable
+  _ <- push rstream =<< (notNull <$> Buffer.fromString strng Encoding.UTF8)
+  _ <- push rstream null -- the end of the stream
+  pure rstream

test/Main.purs

@@ -10,34 +10,27 @@ module Test.Main where
 import Prelude
 
 import Data.Array as Array
-import Data.Either (Either(..))
 import Data.Maybe (Maybe(..))
 import Data.Tuple (Tuple(..), fst)
 import Effect (Effect)
-import Effect.Aff (Error, Milliseconds(..), runAff_)
+import Effect.Aff (Milliseconds(..), launchAff_)
 import Effect.Class (liftEffect)
-import Effect.Class.Console as Console
 import Node.Buffer (Buffer, concat)
 import Node.Buffer as Buffer
 import Node.Encoding (Encoding(..))
 import Node.FS.Stream (createReadStream, createWriteStream)
-import Node.Stream.Aff (end, readAll, readN, readSome, write)
+import Node.Stream.Aff (end, readAll, readN, readSome, toStringUTF8, write)
+import Node.Stream.Aff.Internal (newReadableStringUTF8)
 import Partial.Unsafe (unsafePartial)
 import Test.Spec (describe, it)
 import Test.Spec.Assertions (expectError, shouldEqual)
 import Test.Spec.Reporter (consoleReporter)
-import Test.Spec.Runner (defaultConfig, runSpecT)
-import Unsafe.Coerce (unsafeCoerce)
-
-completion :: Either Error (Effect Unit) -> Effect Unit
-completion = case _ of
-  Left e -> Console.error (unsafeCoerce e)
-  Right f -> f
+import Test.Spec.Runner (defaultConfig, runSpec')
 
 main :: Effect Unit
 main = unsafePartial $ do
-  runAff_ completion do
-    void $ runSpecT (defaultConfig {timeout = Just (Milliseconds 10000.0)}) [consoleReporter] do
+  launchAff_ do
+    runSpec' (defaultConfig {timeout = Just (Milliseconds 20000.0)}) [consoleReporter] do
       describe "Node.Stream.Aff" do
         it "writes and reads" do
           let outfilename = "/tmp/test1.txt"
@@ -56,13 +49,21 @@ main = unsafePartial $ do
           let inputs = input1 <> input2 <> input3
           input :: Buffer <- liftEffect $ concat inputs
           inputSize <- liftEffect $ Buffer.size input
-          shouldEqual (10 * magnitude) inputSize
+          shouldEqual inputSize (10 * magnitude)
         it "writes and closes" do
           let outfilename = "/tmp/test2.txt"
           outfile <- liftEffect $ createWriteStream outfilename
           b <- liftEffect $ Buffer.fromString "test" UTF8
           write outfile [b]
           end outfile
           expectError $ write outfile [b]
+        it "reads from a zero-length Readable" do
+          r <- liftEffect $ newReadableStringUTF8 ""
+          b1 <- toStringUTF8 =<< (fst <$> readSome r)
+          shouldEqual "" b1
+          b2 <- toStringUTF8 =<< (fst <$> readAll r)
+          shouldEqual "" b2
+          b3 <- toStringUTF8 =<< (fst <$> readN r 0)
+          shouldEqual "" b3
 
-    pure (pure unit)
+    pure unit