Clone scala/sys/process/ProcessBuilderImpl.scala

nicolasstucki · nicolasstucki · commit a0c8b82112c6 · 2023-06-14T16:39:28.000+02:00
diff --git a/stdlib-bootstrapped/src/scala/sys/process/ProcessBuilderImpl.scala b/stdlib-bootstrapped/src/scala/sys/process/ProcessBuilderImpl.scala
@@ -0,0 +1,274 @@
+/*
+ * Scala (https://www.scala-lang.org)
+ *
+ * Copyright EPFL and Lightbend, Inc.
+ *
+ * Licensed under Apache License 2.0
+ * (http://www.apache.org/licenses/LICENSE-2.0).
+ *
+ * See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership.
+ */
+
+package scala
+package sys
+package process
+
+import processInternal._
+import Process._
+import BasicIO.{LazilyListed, Streamed, Uncloseable}
+import Uncloseable.protect
+
+import java.io.{FileInputStream, FileOutputStream}
+import java.util.concurrent.LinkedBlockingQueue
+
+import scala.util.control.NonFatal
+
+private[process] trait ProcessBuilderImpl {
+  self: ProcessBuilder.type =>
+
+  private[process] class DaemonBuilder(underlying: ProcessBuilder) extends AbstractBuilder {
+    final def run(io: ProcessIO): Process = underlying.run(io.daemonized())
+  }
+
+  private[process] class Dummy(override val toString: String, exitValue: => Int) extends AbstractBuilder {
+    override def run(io: ProcessIO): Process = new DummyProcess(exitValue)
+    override def canPipeTo = true
+  }
+
+  private[process] class URLInput(url: URL) extends IStreamBuilder(url.openStream(), url.toString)
+  private[process] class FileInput(file: File) extends IStreamBuilder(new FileInputStream(file), file.getAbsolutePath)
+  private[process] class FileOutput(file: File, append: Boolean) extends OStreamBuilder(new FileOutputStream(file, append), file.getAbsolutePath)
+
+  private[process] class OStreamBuilder(
+    stream: => OutputStream,
+    label: String
+  ) extends ThreadBuilder(label, _ writeInput protect(stream)) {
+    override def hasExitValue = false
+  }
+
+  private[process] class IStreamBuilder(
+    stream: => InputStream,
+    label: String
+  ) extends ThreadBuilder(label, _ processOutput protect(stream)) {
+    override def hasExitValue = false
+  }
+
+  private[process] abstract class ThreadBuilder(
+    override val toString: String,
+    runImpl: ProcessIO => Unit
+  ) extends AbstractBuilder {
+
+    override def run(io: ProcessIO): Process = {
+      val success = new LinkedBlockingQueue[Boolean](1)
+      def go(): Unit = {
+        var ok = false
+        try {
+          runImpl(io)
+          ok = true
+        } finally success.put(ok)
+      }
+      val t = Spawn("ThreadProcess", io.daemonizeThreads)(go())
+      new ThreadProcess(t, success)
+    }
+  }
+
+  /** Represents a simple command without any redirection or combination. */
+  private[process] class Simple(p: JProcessBuilder) extends AbstractBuilder {
+    override def run(io: ProcessIO): Process = {
+      import java.lang.ProcessBuilder.Redirect.{INHERIT => Inherit}
+      import io.{daemonizeThreads, processError, processOutput, writeInput}
+
+      val inherit = writeInput eq BasicIO.connectToStdIn
+      if (inherit) p.redirectInput(Inherit)
+
+      val process = p.start() // start the external process
+
+      // spawn threads that process the input, output, and error streams using the functions defined in `io`
+      val inThread =
+        if (inherit || (writeInput eq BasicIO.connectNoOp)) null
+        else Spawn("Simple-input", daemon = true)(writeInput(process.getOutputStream))
+      val outThread = Spawn("Simple-output", daemonizeThreads)(processOutput(process.getInputStream()))
+      val errorThread =
+        if (p.redirectErrorStream) Nil
+        else List(Spawn("Simple-error", daemonizeThreads)(processError(process.getErrorStream())))
+
+      new SimpleProcess(process, inThread, outThread :: errorThread)
+    }
+    override def toString = p.command.toString
+    override def canPipeTo = true
+  }
+
+  private[scala] abstract class AbstractBuilder extends ProcessBuilder with Sink with Source {
+    protected def toSource = this
+    protected def toSink = this
+
+    private[this] val defaultStreamCapacity = 4096
+
+    def #|(other: ProcessBuilder): ProcessBuilder  = {
+      require(other.canPipeTo, "Piping to multiple processes is not supported.")
+      new PipedBuilder(this, other, false)
+    }
+    def #||(other: ProcessBuilder): ProcessBuilder = new OrBuilder(this, other)
+    def #&&(other: ProcessBuilder): ProcessBuilder = new AndBuilder(this, other)
+    def ###(other: ProcessBuilder): ProcessBuilder = new SequenceBuilder(this, other)
+
+    def run(): Process                                          = run(connectInput = false)
+    def run(connectInput: Boolean): Process                     = run(BasicIO.standard(connectInput))
+    def run(log: ProcessLogger): Process                        = run(log, connectInput = false)
+    def run(log: ProcessLogger, connectInput: Boolean): Process = run(BasicIO(connectInput, log))
+
+    def !!                      = slurp(None, withIn = false)
+    def !!(log: ProcessLogger)  = slurp(Some(log), withIn = false)
+    def !!<                     = slurp(None, withIn = true)
+    def !!<(log: ProcessLogger) = slurp(Some(log), withIn = true)
+
+    def lazyLines: LazyList[String]                       = lazyLines(withInput = false, nonZeroException = true, None, defaultStreamCapacity)
+    def lazyLines(log: ProcessLogger): LazyList[String]   = lazyLines(withInput = false, nonZeroException = true, Some(log), defaultStreamCapacity)
+    def lazyLines_! : LazyList[String]                    = lazyLines(withInput = false, nonZeroException = false, None, defaultStreamCapacity)
+    def lazyLines_!(log: ProcessLogger): LazyList[String] = lazyLines(withInput = false, nonZeroException = false, Some(log), defaultStreamCapacity)
+    def lazyLines(capacity: Integer): LazyList[String]                       = lazyLines(withInput = false, nonZeroException = true, None, capacity)
+    def lazyLines(log: ProcessLogger, capacity: Integer): LazyList[String]   = lazyLines(withInput = false, nonZeroException = true, Some(log), capacity)
+    def lazyLines_!(capacity: Integer) : LazyList[String]                    = lazyLines(withInput = false, nonZeroException = false, None, capacity)
+    def lazyLines_!(log: ProcessLogger, capacity: Integer): LazyList[String] = lazyLines(withInput = false, nonZeroException = false, Some(log), capacity)
+
+    @deprecated("internal", since = "2.13.4") def lineStream: Stream[String]                       = lineStream(withInput = false, nonZeroException = true, None, defaultStreamCapacity)
+    @deprecated("internal", since = "2.13.4") def lineStream(log: ProcessLogger): Stream[String]   = lineStream(withInput = false, nonZeroException = true, Some(log), defaultStreamCapacity)
+    @deprecated("internal", since = "2.13.4") def lineStream_! : Stream[String]                    = lineStream(withInput = false, nonZeroException = false, None, defaultStreamCapacity)
+    @deprecated("internal", since = "2.13.4") def lineStream_!(log: ProcessLogger): Stream[String] = lineStream(withInput = false, nonZeroException = false, Some(log), defaultStreamCapacity)
+    @deprecated("internal", since = "2.13.4") def lineStream(capacity: Integer): Stream[String]                       = lineStream(withInput = false, nonZeroException = true, None, capacity)
+    @deprecated("internal", since = "2.13.4") def lineStream(log: ProcessLogger, capacity: Integer): Stream[String]   = lineStream(withInput = false, nonZeroException = true, Some(log), capacity)
+    @deprecated("internal", since = "2.13.4") def lineStream_!(capacity: Integer) : Stream[String]                    = lineStream(withInput = false, nonZeroException = false, None, capacity)
+    @deprecated("internal", since = "2.13.4") def lineStream_!(log: ProcessLogger, capacity: Integer): Stream[String] = lineStream(withInput = false, nonZeroException = false, Some(log), capacity)
+
+    def !                      = run(connectInput = false).exitValue()
+    def !(io: ProcessIO)       = run(io).exitValue()
+    def !(log: ProcessLogger)  = runBuffered(log, connectInput = false)
+    def !<                     = run(connectInput = true).exitValue()
+    def !<(log: ProcessLogger) = runBuffered(log, connectInput = true)
+
+    /** Constructs a new builder which runs this command with all input/output threads marked
+     *  as daemon threads.  This allows the creation of a long running process while still
+     *  allowing the JVM to exit normally.
+     *
+     *  Note: not in the public API because it's not fully baked, but I need the capability
+     *  for fsc.
+     */
+    def daemonized(): ProcessBuilder = new DaemonBuilder(this)
+
+    private[this] def slurp(log: Option[ProcessLogger], withIn: Boolean): String = {
+      val buffer = new StringBuffer
+      val code   = this ! BasicIO(withIn, buffer, log)
+
+      if (code == 0) buffer.toString
+      else scala.sys.error("Nonzero exit value: " + code)
+    }
+
+    private[this] def lazyLines(
+      withInput: Boolean,
+      nonZeroException: Boolean,
+      log: Option[ProcessLogger],
+      capacity: Integer
+    ): LazyList[String] = {
+      val lazilyListed = LazilyListed[String](nonZeroException, capacity)
+      val process      = run(BasicIO(withInput, lazilyListed.process, log))
+
+      // extract done from lazilyListed so that the anonymous function below closes over just the done and not the whole lazilyListed (see https://github.com/scala/bug/issues/12185)
+      val done = lazilyListed.done
+
+      Spawn("LazyLines") {
+        done {
+          try process.exitValue()
+          catch {
+            case NonFatal(_) => -2
+          }
+        }
+      }
+      lazilyListed.lazyList
+    }
+
+    @deprecated("internal", since = "2.13.4")
+    private[this] def lineStream(
+      withInput: Boolean,
+      nonZeroException: Boolean,
+      log: Option[ProcessLogger],
+      capacity: Integer
+    ): Stream[String] = {
+      val streamed = Streamed[String](nonZeroException, capacity)
+      val process  = run(BasicIO(withInput, streamed.process, log))
+
+      Spawn("LineStream")(streamed done process.exitValue())
+      streamed.stream()
+    }
+
+    private[this] def runBuffered(log: ProcessLogger, connectInput: Boolean) =
+      log buffer run(log, connectInput).exitValue()
+
+    def canPipeTo = false
+    def hasExitValue = true
+  }
+
+  private[process] class URLImpl(url: URL) extends URLBuilder with Source {
+    protected def toSource = new URLInput(url)
+  }
+  private[process] class FileImpl(base: File) extends FileBuilder with Sink with Source {
+    protected def toSource = new FileInput(base)
+    protected def toSink   = new FileOutput(base, false)
+
+    def #<<(f: File): ProcessBuilder           = #<<(new FileInput(f))
+    def #<<(u: URL): ProcessBuilder            = #<<(new URLInput(u))
+    def #<<(s: => InputStream): ProcessBuilder = #<<(new IStreamBuilder(s, "<input stream>"))
+    def #<<(b: ProcessBuilder): ProcessBuilder = new PipedBuilder(b, new FileOutput(base, true), false)
+  }
+
+  private[process] abstract class BasicBuilder extends AbstractBuilder {
+    protected[this] def checkNotThis(a: ProcessBuilder) = require(a != this, "Compound process '" + a + "' cannot contain itself.")
+    final def run(io: ProcessIO): Process = {
+      val p = createProcess(io)
+      p.start()
+      p
+    }
+    protected[this] def createProcess(io: ProcessIO): BasicProcess
+  }
+
+  private[process] abstract class SequentialBuilder(
+    a: ProcessBuilder,
+    b: ProcessBuilder,
+    operatorString: String
+  ) extends BasicBuilder {
+
+    checkNotThis(a)
+    checkNotThis(b)
+    override def toString = " ( " + a + " " + operatorString + " " + b + " ) "
+  }
+
+  private[process] class PipedBuilder(
+    first: ProcessBuilder,
+    second: ProcessBuilder,
+    toError: Boolean
+  ) extends SequentialBuilder(first, second, if (toError) "#|!" else "#|") {
+
+    override def createProcess(io: ProcessIO) = new PipedProcesses(first, second, io, toError)
+  }
+
+  private[process] class AndBuilder(
+    first: ProcessBuilder,
+    second: ProcessBuilder
+  ) extends SequentialBuilder(first, second, "#&&") {
+    override def createProcess(io: ProcessIO) = new AndProcess(first, second, io)
+  }
+
+  private[process] class OrBuilder(
+    first: ProcessBuilder,
+    second: ProcessBuilder
+  ) extends SequentialBuilder(first, second, "#||") {
+    override def createProcess(io: ProcessIO) = new OrProcess(first, second, io)
+  }
+
+  private[process] class SequenceBuilder(
+    first: ProcessBuilder,
+    second: ProcessBuilder
+  ) extends SequentialBuilder(first, second, "###") {
+    override def createProcess(io: ProcessIO) = new ProcessSequence(first, second, io)
+  }
+}
