Add multi-client ui to server

This commit makes it possible for the sbt server to render the same ui
to multiple clients. The network client ui should look nearly identical
to the console ui except for the log messages about the experimental
client.

The way that it works is that it associates a ui thread with each
terminal. Whenever a command starts or completes, callbacks are invoked
on the various channels to update their ui state. For example, if there
are two clients and one of them runs compile, then the prompt is changed
from AskUser to Running for the terminal that initiated the command
while the other client remains in the AskUser state. Whenever the client
changes uses ui states, the existing thread is terminated if it is
running and a new thread is begun.

The UITask formalizes this process. It is based on the AskUser class
from older versions of sbt. In fact, there is an AskUserTask which is
very similar. It uses jline to read input from the terminal (which could
be a network terminal). When it gets a line, it submits it to the
CommandExchange and exits. Once the next command is run (which may or
may not be the command it submitted), the ui state will be reset.

The debug, info, warn and error commands should work with the multi
client ui. When run, they set the log level globally, not just for the
client that set the level.

main-command/src/main/scala/sbt/BasicCommandStrings.scala

@@ -235,4 +235,7 @@ $AliasCommand name=
     (ContinuousExecutePrefix + " <command>", continuousDetail)
   def ClearCaches: String = "clearCaches"
   def ClearCachesDetailed: String = "Clears all of sbt's internal caches."
+
+  private[sbt] val networkExecPrefix = "__"
+  private[sbt] val DisconnectNetworkChannel = s"${networkExecPrefix}disconnectNetworkChannel"
 }

main-command/src/main/scala/sbt/BasicCommands.scala

@@ -346,7 +346,13 @@ object BasicCommands {
   private[this] def classpathStrings: Parser[Seq[String]] =
     token(StringBasic.map(s => IO.pathSplit(s).toSeq), "<classpath>")
 
-  def exit: Command = Command.command(TerminateAction, exitBrief, exitBrief)(_ exit true)
+  def exit: Command = Command.command(TerminateAction, exitBrief, exitBrief) { s =>
+    s.source match {
+      case Some(c) if c.channelName.startsWith("network") =>
+        s"${DisconnectNetworkChannel} ${c.channelName}" :: s
+      case _ => s exit true
+    }
+  }
 
   @deprecated("Replaced by BuiltInCommands.continuous", "1.3.0")
   def continuous: Command =

main-command/src/main/scala/sbt/BasicKeys.scala

@@ -13,7 +13,7 @@ import com.github.ghik.silencer.silent
 import sbt.internal.inc.classpath.{ ClassLoaderCache => IncClassLoaderCache }
 import sbt.internal.classpath.ClassLoaderCache
 import sbt.internal.server.ServerHandler
-import sbt.internal.util.AttributeKey
+import sbt.internal.util.{ AttributeKey, Terminal }
 import sbt.librarymanagement.ModuleID
 import sbt.util.Level
 
@@ -35,6 +35,11 @@ object BasicKeys {
     "The function that constructs the command prompt from the current build state.",
     10000
   )
+  val terminalShellPrompt = AttributeKey[(Terminal, State) => String](
+    "new-shell-prompt",
+    "The function that constructs the command prompt from the current build state for a given terminal.",
+    10000
+  )
   @silent val watch =
     AttributeKey[Watched]("watched", "Continuous execution configuration.", 1000)
   val serverPort =

main-command/src/main/scala/sbt/internal/CommandChannel.scala

@@ -9,9 +9,13 @@ package sbt
 package internal
 
 import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.atomic.AtomicReference
 
+import sbt.internal.ui.{ UITask, UserThread }
 import sbt.internal.util.Terminal
 import sbt.protocol.EventMessage
+import sbt.util.Level
+
 import scala.collection.JavaConverters._
 
 /**
@@ -48,6 +52,8 @@ abstract class CommandChannel {
   private[sbt] final def initiateMaintenance(task: String): Unit = {
     maintenance.forEach(q => q.synchronized { q.add(new MaintenanceTask(this, task)); () })
   }
+  private[sbt] def mkUIThread: (State, CommandChannel) => UITask
+  private[sbt] def makeUIThread(state: State): UITask = mkUIThread(state, this)
   final def append(exec: Exec): Boolean = {
     registered.synchronized {
       exec.commandLine.nonEmpty && {
@@ -58,10 +64,29 @@ abstract class CommandChannel {
   }
   def poll: Option[Exec] = Option(commandQueue.poll)
 
+  def prompt(e: ConsolePromptEvent): Unit = userThread.onConsolePromptEvent(e)
+  def unprompt(e: ConsoleUnpromptEvent): Unit = userThread.onConsoleUnpromptEvent(e)
   def publishBytes(bytes: Array[Byte]): Unit
-  def shutdown(): Unit
+  private[sbt] def userThread: UserThread
+  def shutdown(logShutdown: Boolean): Unit = {
+    userThread.stopThread()
+    userThread.close()
+  }
+  @deprecated("Use the variant that takes the logShutdown parameter", "1.4.0")
+  def shutdown(): Unit = shutdown(true)
   def name: String
+  private[this] val level = new AtomicReference[Level.Value](Level.Info)
+  private[sbt] final def setLevel(l: Level.Value): Unit = level.set(l)
+  private[sbt] final def logLevel: Level.Value = level.get
+  private[this] def setLevel(value: Level.Value, cmd: String): Boolean = {
+    level.set(value)
+    append(Exec(cmd, Some(Exec.newExecId), Some(CommandSource(name))))
+  }
   private[sbt] def onCommand: String => Boolean = {
+    case "error" => setLevel(Level.Error, "error")
+    case "debug" => setLevel(Level.Debug, "debug")
+    case "info"  => setLevel(Level.Info, "info")
+    case "warn"  => setLevel(Level.Warn, "warn")
     case cmd =>
       if (cmd.nonEmpty) append(Exec(cmd, Some(Exec.newExecId), Some(CommandSource(name))))
       else false
@@ -89,7 +114,6 @@ case class ConsolePromptEvent(state: State) extends EventMessage
 /*
  * This is a data passed specifically for unprompting local console.
  */
-@deprecated("No longer used", "1.4.0")
 case class ConsoleUnpromptEvent(lastSource: Option[CommandSource]) extends EventMessage
 
 private[internal] class MaintenanceTask(val channel: CommandChannel, val task: String)

main-command/src/main/scala/sbt/internal/ConsoleChannel.scala

@@ -8,76 +8,24 @@
 package sbt
 package internal
 
-import java.io.File
-import java.nio.channels.ClosedChannelException
-import java.util.concurrent.atomic.AtomicReference
-
-import sbt.BasicKeys._
+import sbt.internal.ui.{ UITask, UserThread }
 import sbt.internal.util._
+import sjsonnew.JsonFormat
 
-private[sbt] final class ConsoleChannel(val name: String) extends CommandChannel {
-  private[this] val askUserThread = new AtomicReference[AskUserThread]
-  private[this] def getPrompt(s: State): String = s.get(shellPrompt) match {
-    case Some(pf) => pf(s)
-    case None =>
-      def ansi(s: String): String = if (ConsoleAppender.formatEnabledInEnv) s"$s" else ""
-      s"${ansi(ConsoleAppender.DeleteLine)}> ${ansi(ConsoleAppender.ClearScreenAfterCursor)}"
-  }
-  private[this] class AskUserThread(s: State) extends Thread("ask-user-thread") {
-    private val history = s.get(historyPath).getOrElse(Some(new File(s.baseDir, ".history")))
-    private val prompt = getPrompt(s)
-    private val reader =
-      new FullReader(
-        history,
-        s.combinedParser,
-        LineReader.HandleCONT,
-        Terminal.console,
-      )
-    setDaemon(true)
-    start()
-    override def run(): Unit =
-      try {
-        reader.readLine(prompt) match {
-          case Some(cmd) => append(Exec(cmd, Some(Exec.newExecId), Some(CommandSource(name))))
-          case None =>
-            println("") // Prevents server shutdown log lines from appearing on the prompt line
-            append(Exec("exit", Some(Exec.newExecId), Some(CommandSource(name))))
-        }
-        ()
-      } catch {
-        case _: ClosedChannelException =>
-      } finally askUserThread.synchronized(askUserThread.set(null))
-    def redraw(): Unit = {
-      System.out.print(ConsoleAppender.clearLine(0))
-      System.out.print(ConsoleAppender.ClearScreenAfterCursor)
-      System.out.flush()
-    }
-  }
-  private[this] def makeAskUserThread(s: State): AskUserThread = new AskUserThread(s)
+private[sbt] final class ConsoleChannel(
+    val name: String,
+    override private[sbt] val mkUIThread: (State, CommandChannel) => UITask
+) extends CommandChannel {
 
   def run(s: State): State = s
 
   def publishBytes(bytes: Array[Byte]): Unit = ()
 
-  def prompt(event: ConsolePromptEvent): Unit = {
-    if (Terminal.systemInIsAttached) {
-      askUserThread.synchronized {
-        askUserThread.get match {
-          case null => askUserThread.set(makeAskUserThread(event.state))
-          case t    => t.redraw()
-        }
-      }
-    }
-  }
+  def publishEvent[A: JsonFormat](event: A, execId: Option[String]): Unit = ()
 
-  def shutdown(): Unit = askUserThread.synchronized {
-    askUserThread.get match {
-      case null =>
-      case t if t.isAlive =>
-        t.interrupt()
-        askUserThread.set(null)
-      case _ => ()
-    }
-  }
-  override private[sbt] def terminal = Terminal.console
+  override val userThread: UserThread = new UserThread(this)
+  private[sbt] def terminal = Terminal.console
+}
+private[sbt] object ConsoleChannel {
+  private[sbt] def defaultName = "console0"
 }

main-command/src/main/scala/sbt/internal/client/ServerConnection.scala

@@ -20,12 +20,14 @@ import sbt.internal.util.ReadJsonFromInputStream
 abstract class ServerConnection(connection: Socket) {
 
   private val running = new AtomicBoolean(true)
+  private val closed = new AtomicBoolean(false)
   private val retByte: Byte = '\r'.toByte
   private val delimiter: Byte = '\n'.toByte
 
   private val out = connection.getOutputStream
 
   val thread = new Thread(s"sbt-serverconnection-${connection.getPort}") {
+    setDaemon(true)
     override def run(): Unit = {
       try {
         val in = connection.getInputStream
@@ -67,17 +69,22 @@ abstract class ServerConnection(connection: Socket) {
     writeLine(a)
   }
 
-  def writeLine(a: Array[Byte]): Unit = {
-    def writeEndLine(): Unit = {
-      out.write(retByte.toInt)
-      out.write(delimiter.toInt)
-      out.flush
+  def writeLine(a: Array[Byte]): Unit =
+    try {
+      def writeEndLine(): Unit = {
+        out.write(retByte.toInt)
+        out.write(delimiter.toInt)
+        out.flush
+      }
+      if (a.nonEmpty) {
+        out.write(a)
+      }
+      writeEndLine
+    } catch {
+      case e: IOException =>
+        shutdown()
+        throw e
     }
-    if (a.nonEmpty) {
-      out.write(a)
-    }
-    writeEndLine
-  }
 
   def onRequest(msg: JsonRpcRequestMessage): Unit
   def onResponse(msg: JsonRpcResponseMessage): Unit
@@ -85,10 +92,14 @@ abstract class ServerConnection(connection: Socket) {
 
   def onShutdown(): Unit
 
-  def shutdown(): Unit = {
-    println("Shutting down client connection")
-    running.set(false)
-    out.close()
+  def shutdown(): Unit = if (closed.compareAndSet(false, true)) {
+    if (!running.compareAndSet(true, false)) {
+      System.err.println("\nsbt server connection closed.")
+    }
+    try {
+      out.close()
+      connection.close()
+    } catch { case e: IOException => e.printStackTrace() }
     onShutdown
   }
 

main-command/src/main/scala/sbt/internal/server/Server.scala

@@ -227,7 +227,7 @@ private[sbt] case class ServerConnection(
     socketfile: File,
     pipeName: String,
     bspConnectionFile: File,
-    appConfiguration: AppConfiguration
+    appConfiguration: AppConfiguration,
 ) {
   def shortName: String = {
     connectionType match {

main-command/src/main/scala/sbt/internal/ui/UITask.scala

@@ -0,0 +1,99 @@
+/*
+ * sbt
+ * Copyright 2011 - 2018, Lightbend, Inc.
+ * Copyright 2008 - 2010, Mark Harrah
+ * Licensed under Apache License 2.0 (see LICENSE)
+ */
+
+package sbt.internal.ui
+
+import java.io.File
+import java.nio.channels.ClosedChannelException
+import java.util.concurrent.atomic.AtomicBoolean
+
+import jline.console.history.PersistentHistory
+import sbt.BasicKeys.{ historyPath, terminalShellPrompt }
+import sbt.State
+import sbt.internal.CommandChannel
+import sbt.internal.util.ConsoleAppender.{ ClearPromptLine, ClearScreenAfterCursor, DeleteLine }
+import sbt.internal.util._
+import sbt.internal.util.complete.{ JLineCompletion, Parser }
+
+import scala.annotation.tailrec
+
+private[sbt] trait UITask extends Runnable with AutoCloseable {
+  private[sbt] def channel: CommandChannel
+  private[sbt] def reader: UITask.Reader
+  private[this] final def handleInput(s: Either[String, String]): Boolean = s match {
+    case Left(m)    => channel.onMaintenance(m)
+    case Right(cmd) => channel.onCommand(cmd)
+  }
+  private[this] val isStopped = new AtomicBoolean(false)
+  override def run(): Unit = {
+    @tailrec def impl(): Unit = {
+      val res = reader.readLine()
+      if (!handleInput(res) && !isStopped.get) impl()
+    }
+    try impl()
+    catch { case _: InterruptedException | _: ClosedChannelException => isStopped.set(true) }
+  }
+  override def close(): Unit = isStopped.set(true)
+}
+
+private[sbt] object UITask {
+  trait Reader { def readLine(): Either[String, String] }
+  object Reader {
+    def terminalReader(parser: Parser[_])(
+        terminal: Terminal,
+        state: State
+    ): Reader = {
+      val lineReader = LineReader.createReader(history(state), terminal, terminal.prompt)
+      JLineCompletion.installCustomCompletor(lineReader, parser)
+      () => {
+        val clear = terminal.ansi(ClearPromptLine, "")
+        try {
+          @tailrec def impl(): Either[String, String] = {
+            lineReader.readLine(clear + terminal.prompt.mkPrompt()) match {
+              case null => Left("exit")
+              case s: String =>
+                lineReader.getHistory match {
+                  case p: PersistentHistory =>
+                    p.add(s)
+                    p.flush()
+                  case _ =>
+                }
+                s match {
+                  case ""                                     => impl()
+                  case cmd @ ("shutdown" | "exit" | "cancel") => Left(cmd)
+                  case cmd =>
+                    if (terminal.prompt != Prompt.Batch) terminal.setPrompt(Prompt.Running)
+                    terminal.printStream.write(Int.MinValue)
+                    Right(cmd)
+                }
+            }
+          }
+          impl()
+        } catch {
+          case _: InterruptedException => Right("")
+        } finally lineReader.close()
+      }
+    }
+  }
+  private[this] def history(s: State): Option[File] =
+    s.get(historyPath).getOrElse(Some(new File(s.baseDir, ".history")))
+  private[sbt] def shellPrompt(terminal: Terminal, s: State): String =
+    s.get(terminalShellPrompt) match {
+      case Some(pf) => pf(terminal, s)
+      case None =>
+        def ansi(s: String): String = if (terminal.isAnsiSupported) s"$s" else ""
+        s"${ansi(DeleteLine)}> ${ansi(ClearScreenAfterCursor)}"
+    }
+  private[sbt] class AskUserTask(
+      state: State,
+      override val channel: CommandChannel,
+  ) extends UITask {
+    override private[sbt] def reader: UITask.Reader = {
+      UITask.Reader.terminalReader(state.combinedParser)(channel.terminal, state)
+    }
+  }
+}

main-command/src/main/scala/sbt/internal/ui/UserThread.scala

@@ -0,0 +1,91 @@
+/*
+ * sbt
+ * Copyright 2011 - 2018, Lightbend, Inc.
+ * Copyright 2008 - 2010, Mark Harrah
+ * Licensed under Apache License 2.0 (see LICENSE)
+ */
+
+package sbt.internal
+
+package ui
+
+import java.util.concurrent.atomic.{ AtomicBoolean, AtomicReference }
+import java.util.concurrent.Executors
+
+import sbt.State
+import sbt.internal.util.{ ConsoleAppender, ProgressEvent, ProgressState, Util }
+import sbt.internal.util.Prompt.{ AskUser, Running }
+
+private[sbt] class UserThread(val channel: CommandChannel) extends AutoCloseable {
+  private[this] val uiThread = new AtomicReference[(UITask, Thread)]
+  private[sbt] final def onProgressEvent(pe: ProgressEvent): Unit = {
+    lastProgressEvent.set(pe)
+    ProgressState.updateProgressState(pe, channel.terminal)
+  }
+  private[this] val executor =
+    Executors.newSingleThreadExecutor(r => new Thread(r, s"sbt-$name-ui-thread"))
+  private[this] val lastProgressEvent = new AtomicReference[ProgressEvent]
+  private[this] val isClosed = new AtomicBoolean(false)
+
+  private[sbt] def reset(state: State): Unit = if (!isClosed.get) {
+    uiThread.synchronized {
+      val task = channel.makeUIThread(state)
+      def submit(): Thread = {
+        val thread = new Thread(() => {
+          task.run()
+          uiThread.set(null)
+        }, s"sbt-$name-ui-thread")
+        thread.setDaemon(true)
+        thread.start()
+        uiThread.getAndSet((task, thread)) match {
+          case null   =>
+          case (_, t) => t.interrupt()
+        }
+        thread
+      }
+      uiThread.get match {
+        case null                                  => uiThread.set((task, submit()))
+        case (t, _) if t.getClass == task.getClass =>
+        case (t, thread) =>
+          thread.interrupt()
+          uiThread.set((task, submit()))
+      }
+    }
+    Option(lastProgressEvent.get).foreach(onProgressEvent)
+  }
+
+  private[sbt] def stopThread(): Unit = uiThread.synchronized {
+    uiThread.getAndSet(null) match {
+      case null =>
+      case (t, thread) =>
+        t.close()
+        Util.ignoreResult(thread.interrupt())
+    }
+  }
+
+  private[sbt] def onConsolePromptEvent(consolePromptEvent: ConsolePromptEvent): Unit = {
+    channel.terminal.withPrintStream { ps =>
+      ps.print(ConsoleAppender.ClearScreenAfterCursor)
+      ps.flush()
+    }
+    val state = consolePromptEvent.state
+    terminal.prompt match {
+      case Running => terminal.setPrompt(AskUser(() => UITask.shellPrompt(terminal, state)))
+      case _       =>
+    }
+    onProgressEvent(ProgressEvent("Info", Vector(), None, None, None))
+    reset(state)
+  }
+
+  private[sbt] def onConsoleUnpromptEvent(
+      consoleUnpromptEvent: ConsoleUnpromptEvent
+  ): Unit = {
+    if (consoleUnpromptEvent.lastSource.fold(true)(_.channelName != name)) {
+      terminal.progressState.reset()
+    } else stopThread()
+  }
+
+  override def close(): Unit = if (isClosed.compareAndSet(false, true)) executor.shutdown()
+  private def terminal = channel.terminal
+  private def name: String = channel.name
+}