chiark - git - mdw - tripe-android/blob - admin.scala

   1 /* -*-scala-*-
   2  *
   3  * Managing TrIPE administration connections
   4  *
   5  * (c) 2018 Straylight/Edgeware
   6  */
   7
   8 /*----- Licensing notice --------------------------------------------------*
   9  *
  10  * This file is part of the Trivial IP Encryption (TrIPE) Android app.
  11  *
  12  * TrIPE is free software: you can redistribute it and/or modify it under
  13  * the terms of the GNU General Public License as published by the Free
  14  * Software Foundation; either version 3 of the License, or (at your
  15  * option) any later version.
  16  *
  17  * TrIPE is distributed in the hope that it will be useful, but WITHOUT
  18  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  19  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  20  * for more details.
  21  *
  22  * You should have received a copy of the GNU General Public License
  23  * along with TrIPE.  If not, see <https://www.gnu.org/licenses/>.
  24  */
  25
  26 package uk.org.distorted.tripe; package object admin {
  27
  28 /*----- Imports -----------------------------------------------------------*/
  29
  30 import java.io.{BufferedReader, Reader, Writer};
  31 import java.util.concurrent.locks.{Condition, ReentrantLock => Lock};
  32
  33 import scala.collection.mutable.{HashMap, Publisher};
  34 import scala.concurrent.Channel;
  35 import scala.util.control.Breaks;
  36
  37 import Implicits._;
  38
  39 /*----- Classification of server messages ---------------------------------*/
  40
  41 sealed abstract class Message;
  42
  43 sealed abstract class JobMessage extends Message;
  44 case object JobOK extends JobMessage;
  45 final case class JobInfo(info: Seq[String]) extends JobMessage;
  46 final case class JobFail(err: Seq[String]) extends JobMessage;
  47 case object JobLostConnection extends JobMessage;
  48
  49 final case class BackgroundJobMessage(tag: String, msg: JobMessage)
  50         extends Message;
  51 final case class JobDetached(tag: String) extends Message;
  52
  53 sealed abstract class AsyncMessage extends Message;
  54 final case class Trace(msg: String) extends AsyncMessage;
  55 final case class Warning(err: Seq[String]) extends AsyncMessage;
  56 final case class Notify(note: Seq[String]) extends AsyncMessage;
  57 case object ConnectionLost extends AsyncMessage;
  58
  59 sealed abstract class ServiceMessage extends Message;
  60 final case class ServiceCancel(jobid: String) extends ServiceMessage;
  61 final case class ServiceClaim(svc: String, version: String)
  62         extends ServiceMessage;
  63 final case class ServiceJob(jobid: String, svc: String,
  64                       cmd: String, args: Seq[String])
  65         extends ServiceMessage;
  66
  67 /*----- Main code ---------------------------------------------------------*/
  68
  69 class ConnectionClosed extends Exception;
  70
  71 class ServerFailed(msg: String) extends Exception(msg);
  72
  73 class CommandFailed(val msg: Seq[String]) extends Exception {
  74   override def getMessage(): String =
  75     "%s(%s)".format(getClass.getName, quoteTokens(msg));
  76 }
  77
  78 class ConnectionLostException extends Exception;
  79
  80 class Connection(val in: Reader, val out: Writer)
  81         extends Publisher[AsyncMessage]
  82 {
  83   /* Synchronization.
  84    *
  85    * This class is complicatedly multithreaded.  The following fields must
  86    * only be accessed while the instance is locked.  To prevent deadlocks,
  87    * hold the `Connection' lock before locking any individual `Job' objects.
  88    */
  89
  90   var livep: Boolean = true;            // Is this connection still alive?
  91   var fgjob: Option[this.Job] = None;   // Foreground job, if there is one.
  92   val jobmap = new HashMap[String, this.Job]; // Maps tags to extant jobs.
  93   var bgseq = 0;                        // Next background job tag.
  94
  95   type Pub = Connection;
  96
  97   class Job extends Iterator[Seq[String]] {
  98     private[Connection] val ch = new Channel[JobMessage];
  99     private[this] var nextmsg: Option[JobMessage] = None;
 100
 101     private[this] def fetchNext()
 102       { if (nextmsg == None) nextmsg = Some(ch.read); }
 103     override def hasNext: Boolean = {
 104       fetchNext();
 105       nextmsg match {
 106         case Some(JobOK) => false
 107         case _ => true
 108       }
 109     }
 110     override def next(): Seq[String] = {
 111       fetchNext();
 112       nextmsg match {
 113         case None => ???
 114         case Some(JobOK) => throw new NoSuchElementException
 115         case Some(JobFail(msg)) => throw new CommandFailed(msg)
 116         case Some(JobLostConnection) => throw new ConnectionLostException
 117         case Some(JobInfo(msg)) => nextmsg = None; msg
 118       }
 119     }
 120
 121     def keyvals(): Map[String, String] = {
 122       val b = Map.newBuilder[String, String];
 123       for (line <- this; token <- line) {
 124         token.indexOf('=') match {
 125           case -1 => throw new ServerFailed("missing `=' in key-value list");
 126           case eq =>
 127             val k = token.substring(0, eq);
 128             val v = token.substring(eq + 1);
 129             b += k -> v;
 130         }
 131       }
 132       b.result
 133     }
 134
 135     def traceish(): Seq[(Char, Boolean, String)] = {
 136       val b = Seq.newBuilder[(Char, Boolean, String)];
 137       for (line <- this) line match {
 138         case List(key, desc@_*) =>
 139           val live = if (key.length == 1) false
 140                      else if (key.length == 2 && key(1) == '+') true
 141                      else throw new ServerFailed(
 142                        s"incomprehensible traceish key `$key'");
 143           b += ((key(0), live, desc.mkString(" ")));
 144         case _ => throw new ServerFailed("empty line in traceish output");
 145       }
 146       b.result
 147     }
 148
 149     def expectEmpty() {
 150       if (hasNext) throw new ServerFailed("no output expected");
 151     }
 152
 153     def oneLine(): Seq[String] = {
 154       if (hasNext) {
 155         val line = next();
 156         if (!hasNext) return line;
 157       }
 158       throw new ServerFailed("exactly one line expected");
 159     }
 160   }
 161
 162   def submit(bg: Boolean, toks: String*): this.Job = {
 163     var cmd = toks;
 164 println(";; wait for lock");
 165     synchronized {
 166       if (bg) {
 167         val tag = bgseq formatted "J%05d"; bgseq += 1;
 168         cmd = toks match {
 169           case Seq(cmd, tail@_*) => cmd +: "-background" +: tag +: tail;
 170         }
 171       }
 172 println(";; wait for foreground");
 173       while (livep && fgjob != None) wait();
 174       if (!livep) throw new ConnectionClosed;
 175 println(";; write command");
 176       try { out.write(quoteTokens(cmd)); out.write('\n'); out.flush(); }
 177       catch { case e: Throwable => notify(); throw e; }
 178       val j = new Job;
 179       fgjob = Some(j);
 180       j
 181     }
 182   }
 183
 184   def submit(toks: String*): this.Job = submit(false, toks: _*);
 185
 186   def close() { synchronized { out.close(); } }
 187
 188   /* These two expect the connection lock to be held. */
 189   def foregroundJob: Job =
 190     fgjob.getOrElse { throw new ServerFailed("no foreground job"); }
 191   def releaseForegroundJob() { fgjob = None; notify(); }
 192
 193   def parseServerLine(s: String): Message = nextToken(s) match {
 194     case None => throw new ServerFailed("empty line from server")
 195     case Some(("TRACE", next)) => Trace(s.substring(next))
 196     case Some((code, next)) => (code, splitTokens(s, next)) match {
 197       case ("OK", Seq()) => JobOK
 198       case ("INFO", tail) => JobInfo(tail)
 199       case ("FAIL", tail) => JobFail(tail)
 200       case ("BGDETACH", Seq(tag)) => JobDetached(tag)
 201       case ("BGOK", Seq(tag)) => BackgroundJobMessage(tag, JobOK)
 202       case ("BGINFO", Seq(tag, tail@_*)) =>
 203         BackgroundJobMessage(tag, JobInfo(tail))
 204       case ("BGFAIL", Seq(tag, tail@_*)) =>
 205         BackgroundJobMessage(tag, JobFail(tail))
 206       case ("WARN", tail) => Warning(tail)
 207       case ("NOTE", tail) => Notify(tail)
 208       case ("SVCCLAIM", Seq(svc, ver)) => ServiceClaim(svc, ver)
 209       case ("SVCJOB", Seq(tag, svc, cmd, args@_*)) =>
 210         ServiceJob(tag, svc, cmd, args)
 211       case ("SVCCANCEL", Seq(tag)) => ServiceCancel(tag)
 212       case (_, tail) => throw new ServerFailed(
 213         "incomprehensible line from server: " + quoteTokens(code +: tail))
 214     }
 215   }
 216
 217   def processJobMessage(msg: JobMessage)
 218                        (getjob: (Boolean) => Job) {
 219     synchronized { getjob(msg.isInstanceOf[JobInfo]); }.ch.write(msg);
 220   }
 221
 222   /* Reading lines from the server. */
 223   val readthr = thread("admin reader") {
 224 println(";; readthr running");
 225     val bin = in match {
 226       case br: BufferedReader => br;
 227       case _ => new BufferedReader(in)
 228     }
 229     var line: String = null;
 230
 231     try {
 232 println(";; wait for line");
 233       while ({line = bin.readLine; line != null}) {
 234 println(s";; line: $line");
 235         parseServerLine(line) match {
 236           case JobDetached(tag) => synchronized {
 237             jobmap(tag) = foregroundJob; releaseForegroundJob();
 238           }
 239           case msg: JobMessage => processJobMessage(msg) { keep =>
 240             val j = foregroundJob; if (!keep) releaseForegroundJob(); j
 241           }
 242           case BackgroundJobMessage(tag, msg) =>
 243             processJobMessage(msg) { keep =>
 244               val j = jobmap.getOrElse(tag, throw new ServerFailed(
 245                 s"no job with tag `${tag}'"));
 246               if (!keep) jobmap.remove(tag);
 247               j
 248             }
 249           case msg: AsyncMessage =>
 250             publish(msg);
 251           case _: ServiceMessage =>
 252             ok;
 253         }
 254       }
 255     } catch {
 256       case e: Throwable => e.printStackTrace();
 257     } finally {
 258       synchronized {
 259         livep = false;
 260         for ((_, j) <- jobmap) j.ch.write(JobLostConnection);
 261         fgjob match {
 262           case Some(j) =>
 263             j.ch.write(JobLostConnection);
 264             fgjob = None;
 265             notifyAll();
 266           case None => ok;
 267         }
 268       }
 269       publish(ConnectionLost);
 270       in.close(); out.close();
 271     }
 272   }
 273 }
 274
 275 /*----- That's all, folks -------------------------------------------------*/
 276
 277 }