writeStreamClose

Author: jamesdbrock

src/Node/Stream/Aff.purs

@@ -65,7 +65,7 @@
 -- | Internally the writing functions will call the
 -- | [`writable.write(chunk[, encoding][, callback])`](https://nodejs.org/docs/latest/api/stream.html#writablewritechunk-encoding-callback)
 -- | function on each of the `Buffer`s,
--- | asychronously waiting if there is backpressure from the stream.
+-- | and will asychronously wait if there is “backpressure” from the stream.
 -- |
 -- | The writing functions will complete after all the data is flushed to the
 -- | stream.
@@ -77,14 +77,15 @@
 -- | The canceller argument is an action to perform in the event that
 -- | this `Aff` is cancelled.
 module Node.Stream.Aff
-  ( readAll
+  ( readSome
+  , readSome_
+  , readAll
   , readAll_
   , readN
   , readN_
-  , readSome
-  , readSome_
   , write
   , write_
+  , writableClose
   )
   where
 
@@ -98,14 +99,14 @@ import Data.Either (Either(..))
 import Data.Maybe (Maybe(..))
 import Data.Tuple (Tuple(..))
 import Effect (Effect, untilE)
-import Effect.Aff (effectCanceler, makeAff)
+import Effect.Aff (effectCanceler, makeAff, nonCanceler)
 import Effect.Aff.Class (class MonadAff, liftAff)
 import Effect.Exception (catchException)
 import Node.Buffer (Buffer)
 import Node.Buffer as Buffer
 import Node.Stream (Readable, Writable)
 import Node.Stream as Stream
-import Node.Stream.Aff.Internal (onceDrain, onceEnd, onceError, onceReadable, readable)
+import Node.Stream.Aff.Internal (onceDrain, onceEnd, onceError, onceReadable, readable, writeStreamClose)
 
 
 -- | Wait until there is some data available from the stream, then read it.
@@ -175,7 +176,7 @@ readSome_ r canceller = liftAff <<< makeAff $ \res -> do
   pure $ effectCanceler (canceller r)
 
 
--- | Read all data until the end of the stream. Note that `stdin` will never end.
+-- | Read all data until the end of the stream. Note that __stdin__ will never end.
 readAll
   :: forall m r
    . MonadAff m
@@ -381,3 +382,21 @@ write_ w canceller bs = liftAff <<< makeAff $ \res -> do
   oneWrite
   removeError
   pure $ effectCanceler (canceller w)
+
+-- | Close a `Writable` file stream.
+-- |
+-- | Will complete after the file stream is closed.
+writableClose
+  :: forall m w
+   . MonadAff m
+  => Writable w
+  -> m Unit
+writableClose w = liftAff <<< makeAff $ \res -> do
+
+  removeError <- onceError w $ res <<< Left
+
+  writeStreamClose w do
+    removeError
+    res (Right unit)
+
+  pure nonCanceler

src/Node/Stream/Internal.js

@@ -1,35 +1,27 @@
-import timers from 'node:timers';
-
-export function onceReadable(s) {
-  return f => () => {
-    s.once('readable', f);
-    return () => {s.removeListener('readable', f);};
-  };
+export const onceReadable = s => f => () => {
+  s.once('readable', f);
+  return () => {s.removeListener('readable', f);};
 }
 
-export function onceEnd(s) {
-  return f => () => {
-    s.once('end', f);
-    return () => {s.removeListener('end', f);};
-  };
+export const onceEnd = s => f => () => {
+  s.once('end', f);
+  return () => {s.removeListener('end', f);};
 }
 
-export function onceDrain(s) {
-  return f => () => {
-    s.once('drain', f);
-    return () => {s.removeListener('drain', f);};
-  };
+export const onceDrain = s => f => () => {
+  s.once('drain', f);
+  return () => {s.removeListener('drain', f);};
 }
 
-export function onceError(s) {
-  return f => () => {
-    s.once('error', f);
-    return () => {s.removeListener('error', f);};
-  };
+export const onceError = s => f => () => {
+  s.once('error', f);
+  return () => {s.removeListener('error', f);};
 }
 
-export function readable(s) {
-  return () => {
+export const readable = s => () => {
     return s.readable;
-  }
-}
+}
+
+export const writeStreamClose = s => cb => () => {
+  return s.close(cb);
+}

src/Node/Stream/Internal.purs

@@ -1,9 +1,13 @@
+-- | Maybe the stuff in here should be moved into the
+-- | [__Node.Stream__](https://pursuit.purescript.org/packages/purescript-node-streams/docs/Node.Stream)
+-- | module?
 module Node.Stream.Aff.Internal
   ( onceDrain
   , onceEnd
   , onceError
   , onceReadable
   , readable
+  , writeStreamClose
   )
   where
 
@@ -65,4 +69,14 @@ foreign import onceError
 foreign import readable
   :: forall r
    . Readable r
-  -> Effect Boolean
+  -> Effect Boolean
+
+-- | The [`writeStream.close([callback])`](https://nodejs.org/api/fs.html#writestreamclosecallback)
+-- | function.
+-- |
+-- | Accepts a callback that will be executed once the writeStream is closed.
+foreign import writeStreamClose
+  :: forall w
+   . Writable w
+  -> Effect Unit
+  -> Effect Unit

test/Main.purs

@@ -21,10 +21,10 @@ import Node.Buffer (Buffer, concat)
 import Node.Buffer as Buffer
 import Node.Encoding (Encoding(..))
 import Node.FS.Stream (createReadStream, createWriteStream)
-import Node.Stream.Aff (readAll, readN, readSome, write)
+import Node.Stream.Aff (readAll, readN, readSome, writableClose, write)
 import Partial.Unsafe (unsafePartial)
 import Test.Spec (describe, it)
-import Test.Spec.Assertions (shouldEqual)
+import Test.Spec.Assertions (expectError, shouldEqual)
 import Test.Spec.Reporter (consoleReporter)
 import Test.Spec.Runner (defaultConfig, runSpecT)
 import Unsafe.Coerce (unsafeCoerce)
@@ -57,5 +57,12 @@ main = unsafePartial $ do
           input :: Buffer <- liftEffect $ concat inputs
           inputSize <- liftEffect $ Buffer.size input
           shouldEqual (10 * magnitude) inputSize
+        it "writes and closes" do
+          let outfilename = "/tmp/test2.txt"
+          outfile <- liftEffect $ createWriteStream outfilename
+          b <- liftEffect $ Buffer.fromString "test" UTF8
+          write outfile [b]
+          writableClose outfile
+          expectError $ write outfile [b]
 
     pure (pure unit)