keys.scala, etc.: Make merging public keys have a progress bar.

[tripe-android] / admin.scala

/* -*-scala-*-
 *
 * Managing TrIPE administration connections
 *
 * (c) 2018 Straylight/Edgeware
 */

/*----- Licensing notice --------------------------------------------------*
 *
 * This file is part of the Trivial IP Encryption (TrIPE) Android app.
 *
 * TrIPE is free software: you can redistribute it and/or modify it under
 * the terms of the GNU General Public License as published by the Free
 * Software Foundation; either version 3 of the License, or (at your
 * option) any later version.
 *
 * TrIPE is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
 * for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with TrIPE.  If not, see <https://www.gnu.org/licenses/>.
 */

package uk.org.distorted.tripe; package object admin {

/*----- Imports -----------------------------------------------------------*/

import java.io.{BufferedReader, InputStreamReader, OutputStreamWriter};
import java.util.concurrent.locks.{Condition, ReentrantLock => Lock};

import scala.collection.mutable.HashMap;
import scala.concurrent.Channel;
import scala.util.control.Breaks;

import Implicits._;
import sys.{serverInput, serverOutput};

/*----- Classification of server messages ---------------------------------*/

sealed abstract class Message;

sealed abstract class JobMessage extends Message;
case object JobOK extends JobMessage;
final case class JobInfo(info: Seq[String]) extends JobMessage;
final case class JobFail(err: Seq[String]) extends JobMessage;
case object JobLostConnection extends JobMessage;

final case class BackgroundJobMessage(tag: String, msg: JobMessage)
	extends Message;
final case class JobDetached(tag: String) extends Message;

sealed abstract class AsyncMessage extends Message;
final case class Trace(msg: String) extends AsyncMessage;
final case class Warning(err: Seq[String]) extends AsyncMessage;
final case class Notify(note: Seq[String]) extends AsyncMessage;
case object ConnectionLost extends AsyncMessage;

sealed abstract class ServiceMessage extends Message;
final case class ServiceCancel(jobid: String) extends ServiceMessage;
final case class ServiceClaim(svc: String, version: String)
	extends ServiceMessage;
final case class ServiceJob(jobid: String, svc: String,
		      cmd: String, args: Seq[String])
	extends ServiceMessage;

/*----- Main code ---------------------------------------------------------*/

class ConnectionClosed extends Exception;

class ServerFailed(msg: String) extends Exception(msg);

class CommandFailed(val msg: Seq[String]) extends Exception {
  override def getMessage(): String =
    "%s(%s)".format(getClass.getName, quoteTokens(msg));
}

class ConnectionLostException extends Exception;

object Connection extends Hook[AsyncMessage]
{
  /* Synchronization.
   *
   * This class is complicatedly multithreaded.  The following fields must
   * only be accessed while the instance is locked.  To prevent deadlocks,
   * hold the `Connection' lock before locking any individual `Job' objects.
   */

  private var livep: Boolean = true;	// Is this connection still alive?
  private var fgjob: Option[this.Job] = None; // Foreground job, if there is one.
  private val jobmap = new HashMap[String, this.Job]; // Maps tags to extant jobs.
  private var bgseq = 0;		// Next background job tag.

  private val in = new BufferedReader(new InputStreamReader(serverInput));
  private val out = new OutputStreamWriter(serverOutput);

  type Pub = Connection.type;

  class Job extends Iterator[Seq[String]] {
    private[Connection] val ch = new Channel[JobMessage];
    private[this] var nextmsg: Option[JobMessage] = None;

    private[this] def fetchNext()
      { if (!nextmsg) nextmsg = Some(ch.read); }
    override def hasNext: Boolean = {
      fetchNext();
      nextmsg match {
	case Some(JobOK) => false
	case _ => true
      }
    }
    override def next(): Seq[String] = {
      fetchNext();
      nextmsg match {
	case None => unreachable;
	case Some(JobOK) => throw new NoSuchElementException;
	case Some(JobFail(msg)) => throw new CommandFailed(msg);
	case Some(JobLostConnection) => throw new ConnectionLostException;
	case Some(JobInfo(msg)) => nextmsg = None; msg
      }
    }

    def keyvals(): Map[String, String] = {
      val b = Map.newBuilder[String, String];
      for (line <- this; token <- line) {
	token.indexOf('=') match {
	  case -1 => throw new ServerFailed("missing `=' in key-value list");
	  case eq =>
	    val k = token.substring(0, eq);
	    val v = token.substring(eq + 1);
	    b += k -> v;
	}
      }
      b.result
    }

    def traceish(): Seq[(Char, Boolean, String)] = {
      val b = Seq.newBuilder[(Char, Boolean, String)];
      for (line <- this) line match {
	case List(key, desc@_*) =>
	  val live = if (key.length == 1) false
		     else if (key.length == 2 && key(1) == '+') true
		     else throw new ServerFailed(
		       s"incomprehensible traceish key `$key'");
	  b += ((key(0), live, desc.mkString(" ")));
	case _ => throw new ServerFailed("empty line in traceish output");
      }
      b.result
    }

    def expectEmpty() {
      if (hasNext) throw new ServerFailed("no output expected");
    }

    def oneLine(): Seq[String] = {
      if (hasNext) {
	val line = next();
	if (!hasNext) return line;
      }
      throw new ServerFailed("exactly one line expected");
    }
  }

  def submit(bg: Boolean, toks: String*): this.Job = {
    var cmd = toks;
println(";; wait for lock");
    synchronized {
      if (bg) {
	val tag = bgseq formatted "J%05d"; bgseq += 1;
	cmd = toks match {
	  case Seq(cmd, tail@_*) => cmd +: "-background" +: tag +: tail;
	}
      }
println(";; wait for foreground");
      while (livep && fgjob != None) wait();
      if (!livep) throw new ConnectionClosed;
println(";; write command");
      try { out.write(quoteTokens(cmd)); out.write('\n'); out.flush(); }
      catch { case e: Throwable => notify(); throw e; }
      val j = new Job;
      fgjob = Some(j);
      j
    }
  }

  def submit(toks: String*): this.Job = submit(false, toks: _*);

  /* These two expect the connection lock to be held. */
  def foregroundJob: Job =
    fgjob.getOrElse { throw new ServerFailed("no foreground job"); }
  def releaseForegroundJob() { fgjob = None; notify(); }

  def parseServerLine(s: String): Message = nextToken(s) match {
    case None => throw new ServerFailed("empty line from server")
    case Some(("TRACE", next)) => Trace(s.substring(next))
    case Some((code, next)) => (code, splitTokens(s, next)) match {
      case ("OK", Seq()) => JobOK
      case ("INFO", tail) => JobInfo(tail)
      case ("FAIL", tail) => JobFail(tail)
      case ("BGDETACH", Seq(tag)) => JobDetached(tag)
      case ("BGOK", Seq(tag)) => BackgroundJobMessage(tag, JobOK)
      case ("BGINFO", Seq(tag, tail@_*)) =>
	BackgroundJobMessage(tag, JobInfo(tail))
      case ("BGFAIL", Seq(tag, tail@_*)) =>
	BackgroundJobMessage(tag, JobFail(tail))
      case ("WARN", tail) => Warning(tail)
      case ("NOTE", tail) => Notify(tail)
      case ("SVCCLAIM", Seq(svc, ver)) => ServiceClaim(svc, ver)
      case ("SVCJOB", Seq(tag, svc, cmd, args@_*)) =>
	ServiceJob(tag, svc, cmd, args)
      case ("SVCCANCEL", Seq(tag)) => ServiceCancel(tag)
      case (_, tail) => throw new ServerFailed(
	"incomprehensible line from server: " + quoteTokens(code +: tail))
    }
  }

  def processJobMessage(msg: JobMessage)
		       (getjob: (Boolean) => Job) {
    synchronized { getjob(msg.isInstanceOf[JobInfo]); }.ch.write(msg);
  }

  /* Reading lines from the server. */
  val readthr = thread("admin reader") {
println(";; readthr running");
    val bin = in match {
      case br: BufferedReader => br;
      case _ => new BufferedReader(in)
    }
    var line: String = null;

    try {
println(";; wait for line");
      while ({line = bin.readLine; line != null}) {
println(s";; line: $line");
	parseServerLine(line) match {
	  case JobDetached(tag) => synchronized {
	    jobmap(tag) = foregroundJob; releaseForegroundJob();
	  }
	  case msg: JobMessage => processJobMessage(msg) { keep =>
	    val j = foregroundJob; if (!keep) releaseForegroundJob(); j
	  }
	  case BackgroundJobMessage(tag, msg) =>
	    processJobMessage(msg) { keep =>
	      val j = jobmap.getOrElse(tag, throw new ServerFailed(
		s"no job with tag `${tag}'"));
	      if (!keep) jobmap.remove(tag);
	      j
	    }
	  case msg: AsyncMessage =>
	    callHook(msg);
	  case _: ServiceMessage =>
	    ok;
	}
      }
    } catch {
      case e: Throwable => e.printStackTrace();
    } finally {
      synchronized {
	livep = false;
	for ((_, j) <- jobmap) j.ch.write(JobLostConnection);
	fgjob match {
	  case Some(j) =>
	    j.ch.write(JobLostConnection);
	    fgjob = None;
	    notifyAll();
	  case None => ok;
	}
      }
      callHook(ConnectionLost);
    }
  }
}

/*----- That's all, folks -------------------------------------------------*/

}