Add helper class to manage streaming process output.

Previously, we had fairly complex code for buffering and chunking
process output into individual logging statements. This commit moves
that logic into a separate class that can be tested in isolation to
ensure that we correctly preserve all relevant output from the process.

metals/src/main/scala/scala/meta/internal/metals/BloopInstall.scala

@@ -1,14 +1,12 @@
 package scala.meta.internal.metals
 
 import java.nio.ByteBuffer
-import java.nio.charset.StandardCharsets
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.ScheduledExecutorService
 import java.util.concurrent.TimeUnit
 import com.zaxxer.nuprocess.NuAbstractProcessHandler
 import com.zaxxer.nuprocess.NuProcess
 import com.zaxxer.nuprocess.NuProcessBuilder
-import fansi.ErrorMode
 import scala.concurrent.ExecutionContext
 import scala.concurrent.Future
 import scala.concurrent.Promise
@@ -214,65 +212,37 @@ object BloopInstall {
   private class ProcessHandler() extends NuAbstractProcessHandler {
     var response: Option[CompletableFuture[_]] = None
     val completeProcess = Promise[BloopInstallResult]()
-    val standardOutput = new StringBuilder
-    val errorOutput = new StringBuilder
+    val stdout = new ProcessOutput(line => scribe.info(line))
+    val stderr = new ProcessOutput(line => scribe.error(line))
 
     override def onStart(nuProcess: NuProcess): Unit = {
       nuProcess.closeStdin(false)
     }
 
     override def onExit(statusCode: Int): Unit = {
+      stdout.onProcessExit()
+      stderr.onProcessExit()
       if (!completeProcess.isCompleted) {
         if (statusCode == 0) {
           completeProcess.trySuccess(BloopInstallResult.Installed)
         } else {
           completeProcess.trySuccess(BloopInstallResult.Failed(statusCode))
         }
       }
-      def getLast(output: StringBuilder, fn: String => Unit) {
-        val last = output.toString()
-        if (last.trim().nonEmpty) {
-          fn(last)
-        }
-      }
-      getLast(errorOutput, scribe.error(_))
-      getLast(standardOutput, scribe.info(_))
       scribe.info(s"build tool exit: $statusCode")
       response.foreach(_.cancel(false))
     }
 
     override def onStdout(buffer: ByteBuffer, closed: Boolean): Unit = {
-      log(standardOutput, closed, buffer)(out => scribe.info(out))
+      if (!closed) {
+        stdout.onByteOutput(buffer)
+      }
     }
 
     override def onStderr(buffer: ByteBuffer, closed: Boolean): Unit = {
-      log(errorOutput, closed, buffer)(out => scribe.error(out))
-    }
-
-    private def log(output: StringBuilder, closed: Boolean, buffer: ByteBuffer)(
-        fn: String => Unit
-    ): Unit = {
       if (!closed) {
-        val text = toPlainString(buffer)
-        output.append(text)
-        val lines = output.linesIterator.toList
-        if (output.endsWith("\n")) {
-          lines.foreach(fn)
-          output.clear()
-        } else {
-          lines.take(lines.size - 1).foreach(fn)
-          val last = lines.last
-          output.clear()
-          output.append(last)
-        }
+        stderr.onByteOutput(buffer)
       }
     }
-
-    private def toPlainString(buffer: ByteBuffer): String = {
-      val bytes = new Array[Byte](buffer.remaining())
-      buffer.get(bytes)
-      val ansiString = new String(bytes, StandardCharsets.UTF_8)
-      fansi.Str(ansiString, ErrorMode.Sanitize).plainText
-    }
   }
 }

metals/src/main/scala/scala/meta/internal/metals/ProcessOutput.scala

@@ -0,0 +1,74 @@
+package scala.meta.internal.metals
+
+import java.lang.StringBuilder
+import java.nio.ByteBuffer
+import java.nio.charset.StandardCharsets
+import fansi.ErrorMode
+
+/**
+ * Handles streaming console output from a system process with potential ANSI codes.
+ *
+ * @param onLine The callback handler when the process has printed a single line.
+ *               Guaranteed to have no newline \n characters.
+ */
+class ProcessOutput(onLine: String => Unit) {
+  private var buffer = new StringBuilder()
+
+  /** The process has exited, clears out buffered output. */
+  def onProcessExit(): Unit = {
+    if (buffer.length() > 0) {
+      onLine(buffer.toString())
+      reset()
+    }
+  }
+
+  def onByteOutput(bytes: ByteBuffer): this.type = {
+    onPlainOutput(toPlainString(bytes))
+  }
+
+  def onStringOutput(text: String): this.type = {
+    onPlainOutput(toPlainString(text))
+  }
+
+  private def onPlainOutput(text: String): this.type = {
+    def loop(start: Int): Unit = {
+      val newline = text.indexOf('\n', start)
+      if (newline < 0) {
+        buffer.append(text, start, text.length())
+      } else {
+        onLine(lineSubstring(text, start, newline))
+        loop(newline + 1)
+      }
+    }
+    loop(0)
+    this
+  }
+
+  private def lineSubstring(text: String, from: Int, to: Int): String = {
+    val end =
+      // Convert \r\n line breaks into \n line breaks.
+      if (to > 0 && text.charAt(to - 1) == '\r') to - 1
+      else to
+    if (buffer.length() == 0) text.substring(from, end)
+    else {
+      val line = buffer.append(text, from, to).toString()
+      reset()
+      line
+    }
+  }
+
+  private def reset(): Unit = (
+    buffer = new StringBuilder()
+  )
+
+  private def toPlainString(buffer: ByteBuffer): String = {
+    val bytes = new Array[Byte](buffer.remaining())
+    buffer.get(bytes)
+    val ansiString = new String(bytes, StandardCharsets.UTF_8)
+    toPlainString(ansiString)
+  }
+
+  private def toPlainString(ansiString: String): String = {
+    fansi.Str(ansiString, ErrorMode.Sanitize).plainText
+  }
+}

tests/unit/src/test/scala/tests/ProcessOutputSuite.scala

@@ -0,0 +1,57 @@
+package tests
+
+import scala.meta.internal.metals.ProcessOutput
+import scala.collection.mutable
+
+object ProcessOutputSuite extends BaseSuite {
+  def check(
+      name: String,
+      fn: ProcessOutput => Unit,
+      expected: List[String]
+  ): Unit = {
+    test(name) {
+      var buf = mutable.ListBuffer.empty[String]
+      val output = new ProcessOutput(line => buf += line)
+      fn(output)
+      output.onProcessExit()
+      assertEquals(buf.toList, expected)
+    }
+  }
+
+  check(
+    "flush",
+    _.onStringOutput("a")
+      .onStringOutput("b"),
+    List("ab")
+  )
+
+  check(
+    "loop",
+    _.onStringOutput("a\nb"),
+    List("a", "b")
+  )
+
+  check(
+    "eol",
+    _.onStringOutput("a\n"),
+    List("a")
+  )
+
+  check(
+    "eol2",
+    _.onStringOutput("a\n\n"),
+    List("a", "")
+  )
+
+  check(
+    "ansi",
+    _.onStringOutput(fansi.Color.Blue("blue").toString()),
+    List("blue")
+  )
+
+  check(
+    "windows",
+    _.onStringOutput("a\r\nb"),
+    List("a", "b")
+  )
+}