diff --git a/main-command/src/main/scala/sbt/internal/client/NetworkClient.scala b/main-command/src/main/scala/sbt/internal/client/NetworkClient.scala index d3cd2dc0e..758f8712d 100644 --- a/main-command/src/main/scala/sbt/internal/client/NetworkClient.scala +++ b/main-command/src/main/scala/sbt/internal/client/NetworkClient.scala @@ -10,14 +10,19 @@ package internal package client import java.io.IOException -import java.net.{ URI, Socket, InetAddress } import java.util.UUID import java.util.concurrent.atomic.{ AtomicBoolean, AtomicReference } import scala.collection.mutable.ListBuffer import scala.util.control.NonFatal +import scala.util.{ Success, Failure } import sbt.protocol._ +import sbt.internal.protocol._ +import sbt.internal.langserver.{ LogMessageParams, MessageType, PublishDiagnosticsParams } import sbt.internal.util.{ JLine, StringEvent, ConsoleAppender } import sbt.util.Level +import sbt.io.syntax._ +import sbt.io.IO +import sjsonnew.support.scalajson.unsafe.Converter class NetworkClient(arguments: List[String]) { self => private val channelName = new AtomicReference("_") @@ -28,64 +33,121 @@ class NetworkClient(arguments: List[String]) { self => private val console = ConsoleAppender("thin1") - def usageError = sys.error("Expecting: sbt client 127.0.0.1:port") val connection = init() + start() + // Open server connection based on the portfile def init(): ServerConnection = { - val u = arguments match { - case List(x) => - if (x contains "://") new URI(x) - else new URI("tcp://" + x) - case _ => usageError - } - val host = Option(u.getHost) match { - case None => usageError - case Some(x) => x - } - val port = Option(u.getPort) match { - case None => usageError - case Some(x) if x == -1 => usageError - case Some(x) => x - } - println(s"client on port $port") - val socket = new Socket(InetAddress.getByName(host), port) - new ServerConnection(socket) { - override def onEvent(event: EventMessage): Unit = self.onEvent(event) - override def onLogEntry(event: StringEvent): Unit = self.onLogEntry(event) + val portfile = (file("project") / "target" / "active.json").getAbsoluteFile + if (!portfile.exists) sys.error("server does not seem to be running.") + val (sk, tkn) = ClientSocket.socket(portfile) + val conn = new ServerConnection(sk) { + override def onNotification(msg: JsonRpcNotificationMessage): Unit = self.onNotification(msg) + override def onRequest(msg: JsonRpcRequestMessage): Unit = self.onRequest(msg) + override def onResponse(msg: JsonRpcResponseMessage): Unit = self.onResponse(msg) override def onShutdown(): Unit = { running.set(false) } } + // initiate handshake + val execId = UUID.randomUUID.toString + val initCommand = InitCommand(tkn, Option(execId)) + conn.sendString(Serialization.serializeCommandAsJsonMessage(initCommand)) + conn } - def onLogEntry(event: StringEvent): Unit = { - val level = event.level match { - case "debug" => Level.Debug - case "info" => Level.Info - case "warn" => Level.Warn - case "error" => Level.Error - } - console.appendLog(level, event.message) - } - - def onEvent(event: EventMessage): Unit = - event match { - case e: ChannelAcceptedEvent => - channelName.set(e.channelName) - println(event) - case e: ExecStatusEvent => - status.set(e.status) - // println(event) - e.execId foreach { execId => - if (e.status == "Done" && (pendingExecIds contains execId)) { - lock.synchronized { - pendingExecIds -= execId - } + /** Called on the response for a returning message. */ + def onReturningReponse(msg: JsonRpcResponseMessage): Unit = { + def printResponse(): Unit = { + msg.result match { + case Some(result) => + // ignore result JSON + console.success("completed") + case _ => + msg.error match { + case Some(err) => + // ignore err details + console.appendLog(Level.Error, "completed") + case _ => // ignore } - } - case e => println(e.toString) + } } + printResponse() + } + + def onResponse(msg: JsonRpcResponseMessage): Unit = { + msg.id foreach { + case execId if pendingExecIds contains execId => + onReturningReponse(msg) + lock.synchronized { + pendingExecIds -= execId + } + case _ => + } + } + + def onNotification(msg: JsonRpcNotificationMessage): Unit = { + def splitToMessage: Vector[(Level.Value, String)] = + (msg.method, msg.params) match { + case ("window/logMessage", Some(json)) => + import sbt.internal.langserver.codec.JsonProtocol._ + Converter.fromJson[LogMessageParams](json) match { + case Success(params) => splitLogMessage(params) + case Failure(e) => Vector() + } + case ("textDocument/publishDiagnostics", Some(json)) => + import sbt.internal.langserver.codec.JsonProtocol._ + Converter.fromJson[PublishDiagnosticsParams](json) match { + case Success(params) => splitDiagnostics(params) + case Failure(e) => Vector() + } + case _ => + Vector( + ( + Level.Warn, + s"unknown event: ${msg.method} " + Serialization.compactPrintJsonOpt(msg.params) + ) + ) + } + splitToMessage foreach { + case (level, msg) => console.appendLog(level, msg) + } + } + + def splitLogMessage(params: LogMessageParams): Vector[(Level.Value, String)] = { + val level = messageTypeToLevel(params.`type`) + Vector((level, params.message)) + } + + def messageTypeToLevel(severity: Long): Level.Value = { + severity match { + case MessageType.Error => Level.Error + case MessageType.Warning => Level.Warn + case MessageType.Info => Level.Info + case MessageType.Log => Level.Debug + } + } + + def splitDiagnostics(params: PublishDiagnosticsParams): Vector[(Level.Value, String)] = { + val uri = new URI(params.uri) + val f = IO.toFile(uri) + + params.diagnostics map { d => + val level = d.severity match { + case Some(severity) => messageTypeToLevel(severity) + case _ => Level.Error + } + val line = d.range.start.line + 1 + val offset = d.range.start.character + 1 + val msg = s"$f:$line:$offset: ${d.message}" + (level, msg) + } + } + + def onRequest(msg: JsonRpcRequestMessage): Unit = { + // ignore + } def start(): Unit = { val reader = JLine.simple(None, JLine.HandleCONT, injectThreadSleep = true) @@ -93,12 +155,8 @@ class NetworkClient(arguments: List[String]) { self => reader.readLine("> ", None) match { case Some("exit") => running.set(false) - case Some(s) => - val execId = UUID.randomUUID.toString - publishCommand(ExecCommand(s, execId)) - lock.synchronized { - pendingExecIds += execId - } + case Some(s) if s.trim.nonEmpty => + val execId = sendExecCommand(s) while (pendingExecIds contains execId) { Thread.sleep(100) } @@ -107,10 +165,19 @@ class NetworkClient(arguments: List[String]) { self => } } - def publishCommand(command: CommandMessage): Unit = { - val bytes = Serialization.serializeCommand(command) + def sendExecCommand(commandLine: String): String = { + val execId = UUID.randomUUID.toString + sendCommand(ExecCommand(commandLine, execId)) + lock.synchronized { + pendingExecIds += execId + } + execId + } + + def sendCommand(command: CommandMessage): Unit = { try { - connection.publish(bytes) + val s = Serialization.serializeCommandAsJsonMessage(command) + connection.sendString(s) } catch { case _: IOException => // log.debug(e.getMessage) diff --git a/main-command/src/main/scala/sbt/internal/client/ServerConnection.scala b/main-command/src/main/scala/sbt/internal/client/ServerConnection.scala index 530e117e1..e1950cf33 100644 --- a/main-command/src/main/scala/sbt/internal/client/ServerConnection.scala +++ b/main-command/src/main/scala/sbt/internal/client/ServerConnection.scala @@ -12,11 +12,12 @@ package client import java.net.{ SocketTimeoutException, Socket } import java.util.concurrent.atomic.AtomicBoolean import sbt.protocol._ -import sbt.internal.util.StringEvent +import sbt.internal.protocol._ abstract class ServerConnection(connection: Socket) { private val running = new AtomicBoolean(true) + private val retByte: Byte = '\r'.toByte private val delimiter: Byte = '\n'.toByte private val out = connection.getOutputStream @@ -28,32 +29,63 @@ abstract class ServerConnection(connection: Socket) { val in = connection.getInputStream connection.setSoTimeout(5000) var buffer: Vector[Byte] = Vector.empty - var bytesRead = 0 - while (bytesRead != -1 && running.get) { - try { - bytesRead = in.read(readBuffer) - buffer = buffer ++ readBuffer.toVector.take(bytesRead) - // handle un-framing - var delimPos = buffer.indexOf(delimiter) - while (delimPos > -1) { - val chunk = buffer.take(delimPos) - buffer = buffer.drop(delimPos + 1) + def readFrame: Array[Byte] = { + def getContentLength: Int = { + readLine.drop(16).toInt + } + val l = getContentLength + readLine + readLine + readContentLength(l) + } - Serialization - .deserializeEvent(chunk) - .fold( - { errorDesc => - val s = new String(chunk.toArray, "UTF-8") - println(s"Got invalid chunk from server: $s \n" + errorDesc) - }, - _ match { - case event: EventMessage => onEvent(event) - case event: StringEvent => onLogEntry(event) - } - ) - delimPos = buffer.indexOf(delimiter) + def readLine: String = { + if (buffer.isEmpty) { + val bytesRead = in.read(readBuffer) + if (bytesRead > 0) { + buffer = buffer ++ readBuffer.toVector.take(bytesRead) } + } + val delimPos = buffer.indexOf(delimiter) + if (delimPos > 0) { + val chunk0 = buffer.take(delimPos) + buffer = buffer.drop(delimPos + 1) + // remove \r at the end of line. + val chunk1 = if (chunk0.lastOption contains retByte) chunk0.dropRight(1) else chunk0 + new String(chunk1.toArray, "utf-8") + } else readLine + } + def readContentLength(length: Int): Array[Byte] = { + if (buffer.size < length) { + val bytesRead = in.read(readBuffer) + if (bytesRead > 0) { + buffer = buffer ++ readBuffer.toVector.take(bytesRead) + } + } + if (length <= buffer.size) { + val chunk = buffer.take(length) + buffer = buffer.drop(length) + chunk.toArray + } else readContentLength(length) + } + + while (running.get) { + try { + val frame = readFrame + Serialization + .deserializeJsonMessage(frame) + .fold( + { errorDesc => + val s = new String(frame.toArray, "UTF-8") + println(s"Got invalid chunk from server: $s \n" + errorDesc) + }, + _ match { + case msg: JsonRpcRequestMessage => onRequest(msg) + case msg: JsonRpcResponseMessage => onResponse(msg) + case msg: JsonRpcNotificationMessage => onNotification(msg) + } + ) } catch { case _: SocketTimeoutException => // its ok } @@ -65,14 +97,28 @@ abstract class ServerConnection(connection: Socket) { } thread.start() - def publish(command: Array[Byte]): Unit = { - out.write(command) - out.write(delimiter.toInt) - out.flush() + def sendString(message: String): Unit = { + val a = message.getBytes("UTF-8") + writeLine(s"""Content-Length: ${a.length + 2}""".getBytes("UTF-8")) + writeLine(Array()) + writeLine(a) } - def onEvent(event: EventMessage): Unit - def onLogEntry(event: StringEvent): Unit + def writeLine(a: Array[Byte]): Unit = { + def writeEndLine(): Unit = { + out.write(retByte.toInt) + out.write(delimiter.toInt) + out.flush + } + if (a.nonEmpty) { + out.write(a) + } + writeEndLine + } + + def onRequest(msg: JsonRpcRequestMessage): Unit + def onResponse(msg: JsonRpcResponseMessage): Unit + def onNotification(msg: JsonRpcNotificationMessage): Unit def onShutdown(): Unit diff --git a/protocol/src/main/scala/sbt/protocol/Serialization.scala b/protocol/src/main/scala/sbt/protocol/Serialization.scala index 21d798b80..4d71f512b 100644 --- a/protocol/src/main/scala/sbt/protocol/Serialization.scala +++ b/protocol/src/main/scala/sbt/protocol/Serialization.scala @@ -12,6 +12,7 @@ import sjsonnew.{ JsonFormat, JsonWriter } import sjsonnew.support.scalajson.unsafe.{ Parser, Converter, CompactPrinter } import sjsonnew.shaded.scalajson.ast.unsafe.{ JValue, JObject, JString } import java.nio.ByteBuffer +import java.util.UUID import scala.util.{ Success, Failure } import sbt.internal.util.StringEvent import sbt.internal.protocol.{ @@ -35,6 +36,33 @@ object Serialization { CompactPrinter(json).getBytes("UTF-8") } + private[sbt] def serializeCommandAsJsonMessage(command: CommandMessage): String = { + import sjsonnew.BasicJsonProtocol._ + + command match { + case x: InitCommand => + val execId = x.execId.getOrElse(UUID.randomUUID.toString) + val opt = x.token match { + case Some(t) => + val json: JValue = Converter.toJson[String](t).get + val v = CompactPrinter(json) + s"""{ "token": $v }""" + case None => "{}" + } + s"""{ "jsonrpc": "2.0", "id": "$execId", "method": "initialize", "params": { "initializationOptions": $opt } }""" + case x: ExecCommand => + val execId = x.execId.getOrElse(UUID.randomUUID.toString) + val json: JValue = Converter.toJson[String](x.commandLine).get + val v = CompactPrinter(json) + s"""{ "jsonrpc": "2.0", "id": "$execId", "method": "sbt/exec", "params": { "commandLine": $v } }""" + case x: SettingQuery => + val execId = UUID.randomUUID.toString + val json: JValue = Converter.toJson[String](x.setting).get + val v = CompactPrinter(json) + s"""{ "jsonrpc": "2.0", "id": "$execId", "method": "sbt/setting", "params": { "setting": $v } }""" + } + } + def serializeEventMessage(event: EventMessage): Array[Byte] = { import codec.JsonProtocol._ val json: JValue = Converter.toJson[EventMessage](event).get @@ -141,18 +169,25 @@ object Serialization { private[sbt] def deserializeJsonMessage(bytes: Seq[Byte]): Either[String, JsonRpcMessage] = { val buffer = ByteBuffer.wrap(bytes.toArray) Parser.parseFromByteBuffer(buffer) match { - case Success(json) => + case Success(json @ JObject(fields)) => import sbt.internal.protocol.codec.JsonRPCProtocol._ - Converter.fromJson[JsonRpcRequestMessage](json) match { - case Success(request) if (request.id.nonEmpty) => Right(request) - case Failure(e) => throw e - case _ => { + if ((fields find { _.field == "method" }).isDefined) { + if ((fields find { _.field == "id" }).isDefined) + Converter.fromJson[JsonRpcRequestMessage](json) match { + case Success(request) => Right(request) + case Failure(e) => Left(s"Conversion error: ${e.getMessage}") + } else Converter.fromJson[JsonRpcNotificationMessage](json) match { case Success(notification) => Right(notification) - case Failure(e) => throw e + case Failure(e) => Left(s"Conversion error: ${e.getMessage}") } + } else + Converter.fromJson[JsonRpcResponseMessage](json) match { + case Success(res) => Right(res) + case Failure(e) => Left(s"Conversion error: ${e.getMessage}") } - } + case Success(json) => + Left(s"Expected JSON object but found $json") case Failure(e) => Left(s"Parse error: ${e.getMessage}") }