src/Pure/Tools/server.scala

explicit Server.Context with output channels (concurrent write);
support for Logger and Progress;

/*  Title:      Pure/Tools/server.scala
    Author:     Makarius

Resident Isabelle servers.

Message formats:
  - short message (single line):
      NAME ARGUMENT
  - long message (multiple lines):
      BYTE_LENGTH
      NAME ARGUMENT

Argument formats:
  - Unit as empty string
  - XML.Elem in YXML notation
  - JSON.T in standard notation
*/

package isabelle


import java.io.{BufferedInputStream, BufferedOutputStream, InputStreamReader, OutputStreamWriter,
  IOException}
import java.net.{Socket, SocketException, SocketTimeoutException, ServerSocket, InetAddress}


object Server
{
  /* message argument */

  object Argument
  {
    def split(msg: String): (String, String) =
    {
      val name =
        msg.takeWhile(c => Symbol.is_ascii_letter(c) || Symbol.is_ascii_letdig(c) || c == '.')
      val argument = msg.substring(name.length).dropWhile(Symbol.is_ascii_blank(_))
      (name, argument)
    }

    def print(arg: Any): String =
      arg match {
        case () => ""
        case t: XML.Elem => YXML.string_of_tree(t)
        case t: JSON.T => JSON.Format(t)
      }

    def parse(argument: String): Any =
      if (argument == "") ()
      else if (YXML.detect_elem(argument)) YXML.parse_elem(argument)
      else JSON.parse(argument, strict = false)

    def unapply(argument: String): Option[Any] =
      try { Some(parse(argument)) }
      catch { case ERROR(_) => None }
  }


  /* input command */

  object Command
  {
    type T = PartialFunction[(Context, Any), Any]

    private val table: Map[String, T] =
      Map(
        "echo" -> { case (_, t) => t },
        "help" -> { case (_, ()) => table.keySet.toList.sorted },
        "shutdown" -> { case (context, ()) => context.shutdown(); () })

    def unapply(name: String): Option[T] = table.get(name)
  }


  /* output reply */

  object Reply extends Enumeration
  {
    val OK, ERROR, NOTE = Value

    def unapply(msg: String): Option[(Reply.Value, Any)] =
    {
      if (msg == "") None
      else {
        val (name, argument) = Argument.split(msg)
        for {
          reply <-
            try { Some(withName(name)) }
            catch { case _: NoSuchElementException => None }
          arg <- Argument.unapply(argument)
        } yield (reply, arg)
      }
    }
  }


  /* socket connection */

  object Connection
  {
    def apply(socket: Socket): Connection =
      new Connection(socket)
  }

  class Connection private(socket: Socket)
  {
    override def toString: String = socket.toString

    def close() { socket.close }

    def set_timeout(t: Time) { socket.setSoTimeout(t.ms.toInt) }

    private val in = new BufferedInputStream(socket.getInputStream)
    private val out = new BufferedOutputStream(socket.getOutputStream)
    private val out_lock: AnyRef = new Object

    def tty_loop(interrupt: Option[() => Unit] = None): TTY_Loop =
      new TTY_Loop(
        new OutputStreamWriter(out),
        new InputStreamReader(in),
        writer_lock = out_lock,
        interrupt = interrupt)

    def read_message(): Option[String] =
      try {
        Bytes.read_line(in).map(_.text) match {
          case Some(Value.Int(n)) =>
            Bytes.read_block(in, n).map(bytes => Library.trim_line(bytes.text))
          case res => res
        }
      }
      catch { case _: SocketException => None }

    def write_message(msg: String): Unit = out_lock.synchronized
    {
      val b = UTF8.bytes(msg)
      if (b.length > 100 || b.contains(10)) {
        out.write(UTF8.bytes((b.length + 1).toString))
        out.write(10)
      }
      out.write(b)
      out.write(10)
      try { out.flush() } catch { case _: SocketException => }
    }

    def reply(r: Server.Reply.Value, arg: Any)
    {
      val argument = Argument.print(arg)
      write_message(if (argument == "") r.toString else r.toString + " " + argument)
    }

    def reply_ok(arg: Any) { reply(Server.Reply.OK, arg) }
    def reply_error(arg: Any) { reply(Server.Reply.ERROR, arg) }
    def reply_error_message(message: String, more: (String, JSON.T)*): Unit =
      reply_error(Map("message" -> message) ++ more)

    def notify(arg: Any) { reply(Server.Reply.NOTE, arg) }
    def notify_message(kind: String, msg: String, more: (String, JSON.T)*): Unit =
      notify(Map(Markup.KIND -> kind, "message" -> msg) ++ more)
  }


  /* context with output channels */

  class Context private[Server](server: Server, connection: Connection)
  {
    context =>

    def shutdown() { server.close() }

    def message(kind: String, msg: String, more: (String, JSON.T)*): Unit =
      connection.notify_message(kind, msg, more:_*)
    def writeln(msg: String, more: (String, JSON.T)*): Unit = message(Markup.WRITELN, msg, more:_*)
    def warning(msg: String, more: (String, JSON.T)*): Unit = message(Markup.WARNING, msg, more:_*)
    def error_message(msg: String, more: (String, JSON.T)*): Unit =
      message(Markup.ERROR_MESSAGE, msg, more:_*)

    val logger: Connection_Logger = new Connection_Logger(context)
    def progress(): Connection_Progress = new Connection_Progress(context)

    override def toString: String = connection.toString
  }

  class Connection_Logger private[Server](context: Context) extends Logger
  {
    def apply(msg: => String): Unit = context.message(Markup.LOGGER, msg)

    override def toString: String = context.toString
  }

  class Connection_Progress private[Server](context: Context) extends Progress
  {
    override def echo(msg: String): Unit = context.writeln(msg)
    override def echo_warning(msg: String): Unit = context.warning(msg)
    override def echo_error_message(msg: String): Unit = context.error_message(msg)
    override def theory(session: String, theory: String): Unit =
      context.writeln(session + ": theory " + theory, "session" -> session, "theory" -> theory)

    @volatile private var is_stopped = false
    override def stopped: Boolean = is_stopped
    def stop { is_stopped = true }

    override def toString: String = context.toString
  }


  /* server info */

  sealed case class Info(name: String, port: Int, password: String)
  {
    override def toString: String =
      "server " + quote(name) + " = " + print(port, password)

    def connection(): Connection =
    {
      val connection = Connection(new Socket(InetAddress.getByName("127.0.0.1"), port))
      connection.write_message(password)
      connection
    }

    def active(): Boolean =
      try {
        using(connection())(connection =>
          {
            connection.set_timeout(Time.seconds(2.0))
            connection.read_message() == Some(Reply.OK.toString)
          })
      }
      catch {
        case _: IOException => false
        case _: SocketException => false
        case _: SocketTimeoutException => false
      }
  }


  /* per-user servers */

  val default_name = "isabelle"

  def print(port: Int, password: String): String =
    "127.0.0.1:" + port + " (password " + quote(password) + ")"

  object Data
  {
    val database = Path.explode("$ISABELLE_HOME_USER/servers.db")

    val name = SQL.Column.string("name").make_primary_key
    val port = SQL.Column.int("port")
    val password = SQL.Column.string("password")
    val table = SQL.Table("isabelle_servers", List(name, port, password))
  }

  def list(db: SQLite.Database): List[Info] =
    if (db.tables.contains(Data.table.name)) {
      db.using_statement(Data.table.select())(stmt =>
        stmt.execute_query().iterator(res =>
          Info(
            res.string(Data.name),
            res.int(Data.port),
            res.string(Data.password))).toList.sortBy(_.name))
    }
    else Nil

  def find(db: SQLite.Database, name: String): Option[Info] =
    list(db).find(server_info => server_info.name == name && server_info.active)

  def init(name: String = default_name, port: Int = 0, existing_server: Boolean = false)
    : (Info, Option[Server]) =
  {
    using(SQLite.open_database(Data.database))(db =>
      {
        db.transaction {
          Isabelle_System.bash("chmod 600 " + File.bash_path(Data.database)).check
          db.create_table(Data.table)
          list(db).filterNot(_.active).foreach(server_info =>
            db.using_statement(Data.table.delete(Data.name.where_equal(server_info.name)))(
              _.execute))
        }
        db.transaction {
          find(db, name) match {
            case Some(server_info) => (server_info, None)
            case None =>
              if (existing_server) error("Isabelle server " + quote(name) + " not running")

              val server = new Server(port)
              val server_info = Info(name, server.port, server.password)

              db.using_statement(Data.table.delete(Data.name.where_equal(name)))(_.execute)
              db.using_statement(Data.table.insert())(stmt =>
              {
                stmt.string(1) = server_info.name
                stmt.int(2) = server_info.port
                stmt.string(3) = server_info.password
                stmt.execute()
              })

              server.start
              (server_info, Some(server))
          }
        }
      })
  }

  def exit(name: String = default_name): Boolean =
  {
    using(SQLite.open_database(Data.database))(db =>
      db.transaction {
        find(db, name) match {
          case Some(server_info) =>
            using(server_info.connection())(_.write_message("shutdown"))
            while(server_info.active) { Thread.sleep(50) }
            true
          case None => false
        }
      })
  }


  /* Isabelle tool wrapper */

  val isabelle_tool =
    Isabelle_Tool("server", "manage resident Isabelle servers", args =>
    {
      var console = false
      var operation_list = false
      var operation_exit = false
      var name = default_name
      var port = 0
      var existing_server = false

      val getopts =
        Getopts("""
Usage: isabelle server [OPTIONS]

  Options are:
    -C           console interaction with specified server
    -L           list servers (exclusive operation)
    -X           exit specified server (exclusive operation)
    -n NAME      explicit server name (default: """ + default_name + """)
    -p PORT      explicit server port
    -s           assume existing server, no implicit startup

  Manage resident Isabelle servers.
""",
          "C" -> (_ => console = true),
          "L" -> (_ => operation_list = true),
          "X" -> (_ => operation_exit = true),
          "n:" -> (arg => name = arg),
          "p:" -> (arg => port = Value.Int.parse(arg)),
          "s" -> (_ => existing_server = true))

      val more_args = getopts(args)
      if (more_args.nonEmpty) getopts.usage()

      if (operation_list) {
        for {
          server_info <- using(SQLite.open_database(Data.database))(list(_))
          if server_info.active
        } Output.writeln(server_info.toString, stdout = true)
      }
      else if (operation_exit) {
        val ok = Server.exit(name)
        sys.exit(if (ok) 0 else 1)
      }
      else {
        val (server_info, server) = init(name, port = port, existing_server = existing_server)
        Output.writeln(server_info.toString, stdout = true)
        if (console) {
          using(server_info.connection())(connection => connection.tty_loop().join)
        }
        server.foreach(_.join)
      }
    })
}

class Server private(_port: Int)
{
  server =>

  private val server_socket = new ServerSocket(_port, 50, InetAddress.getByName("127.0.0.1"))

  def close() { server_socket.close }

  def port: Int = server_socket.getLocalPort
  val password: String = Library.UUID()

  override def toString: String = Server.print(port, password)

  private def handle(connection: Server.Connection)
  {
    val context = new Server.Context(server, connection)

    connection.read_message() match {
      case Some(msg) if msg == password =>
        connection.reply_ok(())
        var finished = false
        while (!finished) {
          connection.read_message() match {
            case None => finished = true
            case Some("") => context.writeln("Command 'help' provides list of commands")
            case Some(msg) =>
              val (name, argument) = Server.Argument.split(msg)
              name match {
                case Server.Command(cmd) =>
                  argument match {
                    case Server.Argument(arg) =>
                      if (cmd.isDefinedAt((context, arg))) {
                        Exn.capture { cmd((context, arg)) } match {
                          case Exn.Res(res) => connection.reply_ok(res)
                          case Exn.Exn(ERROR(msg)) => connection.reply_error(msg)
                          case Exn.Exn(exn) => throw exn
                        }
                      }
                      else {
                        connection.reply_error_message(
                          "Bad argument for command " + Library.single_quote(name),
                          "argument" -> argument)
                      }
                    case _ =>
                      connection.reply_error_message(
                        "Malformed argument for command " + Library.single_quote(name),
                        "argument" -> argument)
                  }
                case _ => connection.reply_error("Bad command " + Library.single_quote(name))
              }
          }
        }
      case _ =>
    }
  }

  private lazy val server_thread: Thread =
    Standard_Thread.fork("server") {
      var finished = false
      while (!finished) {
        Exn.capture(server_socket.accept) match {
          case Exn.Res(socket) =>
            Standard_Thread.fork("server_connection")
              { using(Server.Connection(socket))(handle(_)) }
          case Exn.Exn(_) => finished = true
        }
      }
    }

  def start { server_thread }

  def join { server_thread.join; close() }
}