Make all process operators interruptible

.sbtopts

@@ -0,0 +1,4 @@
+-J-Xmx4G
+-J-XX:MaxMetaspaceSize=1G
+-J-Dfile.encoding=UTF-8
+-J-Dsbt.io.implicit.relative.glob.conversion=allow

project/BuildHelper.scala

@@ -80,6 +80,8 @@ object BuildHelper {
   def stdSettings(prjName: String) = Seq(
     name := s"$prjName",
     fork := true,
+    run / baseDirectory := file("."),
+    Test / baseDirectory := file("."),
     crossScalaVersions := Seq(Scala212, Scala213),
     ThisBuild / scalaVersion := Scala213,
     scalacOptions := stdOptions ++ extraOptions(scalaVersion.value),

zio-process/js/src/main/scala/zio/process/CommandPlatformSpecific.scala

@@ -17,7 +17,7 @@ package zio.process
 
 import FilePlatformSpecific._
 import ProcessPlatformSpecific._
-import zio.{ NonEmptyChunk, ZIO }
+import zio._
 import scala.annotation.nowarn
 import scala.scalajs.js
 import js.JSConverters._
@@ -38,7 +38,7 @@ private[process] trait CommandPlatformSpecific {
       stdinStream <- c.stdin match {
                        case FromStream(stream, _) =>
                          stream.toInputStream.map(Some(_))
-                       case JavaStream(in, _)     => ZIO.succeed(Some(in))
+                       case JavaStream(in, _)     => ZIO.some(in)
                        case _                     => ZIO.none
                      }
       promise     <- ZIO.attempt {
@@ -97,7 +97,11 @@ private[process] trait CommandPlatformSpecific {
                                  process.stdin.write(new js.typedarray.Uint8Array(arr.take(read).map(_.toShort).toJSArray))
                                )
                              else if (read == -1)
-                               if (process.stdin != null) if (!process.stdin.writableEnded) { process.stdin.end(); () }
+                               if (process.stdin != null && !process.stdin.writableEnded) {
+                                 process.stdin.end()
+                                 ()
+                               }
+
                            }
 
                            write(true)
@@ -113,7 +117,10 @@ private[process] trait CommandPlatformSpecific {
                                    "end",
                                    { () =>
                                      connectStdin(s)
-                                     if (process.stdin != null) if (!process.stdin.writableEnded) process.stdin.end()
+
+                                     if (process.stdin != null && !process.stdin.writableEnded)
+                                       process.stdin.end()
+
                                      resolve(zioProcess)
                                    }
                                  )
@@ -122,7 +129,9 @@ private[process] trait CommandPlatformSpecific {
                              process.on("spawn", () => connectStdin(s))
                              in.on(
                                "end",
-                               () => if (process.stdin != null) if (!process.stdin.writableEnded) process.stdin.end()
+                               () =>
+                                 if (process.stdin != null && !process.stdin.writableEnded)
+                                   process.stdin.end()
                              )
                              val node = js.Dynamic.global.require("process")
                              process.on("spawn", () => node.nextTick(() => resolve(zioProcess)))

zio-process/js/src/main/scala/zio/process/Constructors.scala

@@ -20,10 +20,13 @@ import zio.stream.ZStream
 
 import java.io.OutputStream
 import zio.Trace
+
 import java.io.IOException
 import zio.ZIO
 import zio.Chunk
 import zio.Scope
+import zio.process.ProcessPlatformSpecific.JProcess
+
 import java.io.InputStream
 
 private[process] object Constructors {
@@ -52,19 +55,14 @@ private[process] object Constructors {
             val bytes = byteChunk.toArray
             out.write(bytes)
             bytesWritten + bytes.length
-          }.refineOrDie { case e: IOException =>
-            e
-          }
+          }.refineToOrDie[IOException]
         }
       }
     }
 
   def zsink(outputStream: OutputStream): ZSink[Any, IOException, Byte, Byte, Long] = fromOutputStream(outputStream)
 
-  /**
-   * Creates a stream from a `java.io.InputStream`
-   */
-  def fromInputStream(
+  def fromProcessInputStream(process: JProcess)(
     is: => InputStream,
     chunkSize: => Int = ZStream.DefaultChunkSize
   )(implicit trace: Trace): ZStream[Any, IOException, Byte] =
@@ -73,8 +71,10 @@ private[process] object Constructors {
         for {
           bufArray  <- ZIO.succeed(Array.ofDim[Byte](chunkSize))
           bytesRead <- ZIO
-                         .attemptBlockingCancelable(is.read(bufArray))(ZIO.succeed(is.close()))
-                         .refineToOrDie[java.io.IOException]
+                         .attemptBlockingCancelable(is.read(bufArray))(
+                           ZIO.succeed(is.close()).ensuring(ZIO.attemptBlocking(process.kill()).ignore)
+                         )
+                         .refineToOrDie[IOException]
                          .asSomeError
           bytes     <- if (bytesRead < 0)
                          ZIO.fail(None)

zio-process/js/src/main/scala/zio/process/FilePlatformSpecific.scala

@@ -6,6 +6,8 @@ private[process] object FilePlatformSpecific {
   type File = String
   type Path = String
 
+  def fileOf(file: String): File = file
+
   def getAbsolute(file: File): js.Any = {
     val path   = js.Dynamic.global.require("path")
     val nodejs = js.Dynamic.global.require("process")

zio-process/js/src/main/scala/zio/process/ProcessInput.scala

@@ -16,7 +16,7 @@
 package zio.process
 
 import zio.stream.ZStream
-import zio.{ Chunk, Queue }
+import zio._
 import FilePlatformSpecific._
 import java.io.ByteArrayInputStream
 import java.nio.charset.{ Charset, StandardCharsets }
@@ -41,7 +41,7 @@ object ProcessInput {
     ProcessInput.JavaStream(
       ProcessPlatformSpecific.JSInputStream(
         fs.createReadStream(FilePlatformSpecific.getAbsolute(file)).asInstanceOf[JS.Readable],
-        true
+        pause = true
       ),
       flushChunksEagerly = false
     )

zio-process/js/src/main/scala/zio/process/ProcessPlatformSpecific.scala

@@ -15,12 +15,11 @@
  */
 package zio.process
 
-import zio.ZIO
+import zio._
 
 import java.io.{ InputStream, OutputStream }
 import scala.scalajs.js.typedarray.Uint8Array
 import scala.scalajs.js
-import zio.Chunk
 import js.JSConverters._
 
 private[process] trait ProcessPlatformSpecific extends ProcessInterface { self: Process =>
@@ -29,7 +28,13 @@ private[process] trait ProcessPlatformSpecific extends ProcessInterface { self:
 
   private var killed = false
 
-  protected def waitForUnsafe: Int = self.process.exitCode
+  def waitFor: IO[CommandError, ExitCode] =
+    ProcessPlatformSpecific.wait(stdoutInternal).map(_ => ExitCode(self.process.exitCode)).refineOrDie {
+      case CommandThrowable.IOError(e) => e
+    }
+
+  protected def waitForUnsafe: Int =
+    self.process.exitCode
 
   protected def isAliveUnsafe: Boolean = !killed
   protected def destroyUnsafe(): Unit = {
@@ -139,6 +144,7 @@ private[process] object ProcessPlatformSpecific {
       }
 
     }
+
     override def readAllBytes(): Array[Byte] = {
       val arr = buffer.toArray
       buffer = Chunk.empty

zio-process/jvm/src/main/scala/zio/process/Constructors.scala

@@ -15,24 +15,42 @@
  */
 package zio.process
 
+import zio.process.ProcessPlatformSpecific.JProcess
 import zio.stream.ZSink
 import zio.stream.ZStream
 
 import java.io.OutputStream
 import java.io.InputStream
-import zio.Trace
+import zio._
+
 import java.io.IOException
 
 private[process] object Constructors {
 
   def zsink(outputStream: OutputStream) = ZSink.fromOutputStream(outputStream)
 
-  /**
-   * Creates a stream from a `java.io.InputStream`
-   */
-  def fromInputStream(
+  def fromProcessInputStream(process: JProcess)(
     is: => InputStream,
     chunkSize: => Int = ZStream.DefaultChunkSize
-  )(implicit trace: Trace): ZStream[Any, IOException, Byte] = ZStream.fromInputStream(is, chunkSize)
+  )(implicit trace: Trace): ZStream[Any, IOException, Byte] =
+    ZStream.succeed((is, chunkSize)).flatMap { case (is, chunkSize) =>
+      ZStream.repeatZIOChunkOption {
+        for {
+          bufArray  <- ZIO.succeed(Array.ofDim[Byte](chunkSize))
+          bytesRead <- ZIO
+                         .attemptBlockingCancelable(is.read(bufArray))(ZIO.attemptBlocking(process.destroy()).ignore)
+                         .refineToOrDie[IOException]
+                         .asSomeError
+          bytes     <- if (bytesRead < 0)
+                         ZIO.fail(None)
+                       else if (bytesRead == 0)
+                         ZIO.succeed(Chunk.empty)
+                       else if (bytesRead < chunkSize)
+                         ZIO.succeed(Chunk.fromArray(bufArray).take(bytesRead))
+                       else
+                         ZIO.succeed(Chunk.fromArray(bufArray))
+        } yield bytes
+      }
+    }
 
 }

zio-process/jvm/src/main/scala/zio/process/FilePlatformSpecific.scala

@@ -1,7 +1,9 @@
 package zio.process
 
-object FilePlatformSpecific {
+private[process] object FilePlatformSpecific {
   type File = java.io.File
 
-  def exists(file: File) = file.exists()
+  def fileOf(file: String): File = new File(file)
+
+  def exists(file: File): Boolean = file.exists()
 }

zio-process/jvm/src/main/scala/zio/process/ProcessInput.scala

@@ -16,7 +16,7 @@
 package zio.process
 
 import zio.stream.ZStream
-import zio.{ Chunk, Queue }
+import zio._
 
 import java.io.ByteArrayInputStream
 import java.nio.charset.{ Charset, StandardCharsets }