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 Magic._;
  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 case class JobInfo(info: Seq[String]) extends JobMessage;
  46 case class JobFail(err: Seq[String]) extends JobMessage;
  47 case object JobLostConnection extends JobMessage;
  48
  49 case class BackgroundJobMessage(tag: String, msg: JobMessage)
  50         extends Message;
  51 case class JobDetached(tag: String) extends Message;
  52
  53 sealed abstract class AsyncMessage extends Message;
  54 case class Trace(msg: String) extends AsyncMessage;
  55 case class Warning(err: Seq[String]) extends AsyncMessage;
  56 case class Notify(note: Seq[String]) extends AsyncMessage;
  57 case object ConnectionLost extends AsyncMessage;
  58
  59 sealed abstract class ServiceMessage extends Message;
  60 case class ServiceCancel(jobid: String) extends ServiceMessage;
  61 case class ServiceClaim(svc: String, version: String)
  62         extends ServiceMessage;
  63 case class ServiceJob(jobid: String, svc: String,
  64                       cmd: String, args: Seq[String])
  65         extends ServiceMessage;
  66
  67 /*----- Main code ---------------------------------------------------------*/
  68
  69 object Connection {
  70 }
  71
  72 class ConnectionClosed extends Exception;
  73
  74 class ServerFailed(msg: String) extends Exception(msg);
  75
  76 class CommandFailed(val msg: Seq[String]) extends Exception {
  77   override def getMessage(): String =
  78     "%s(%s)".format(getClass.getName, quoteTokens(msg));
  79 }
  80
  81 class ConnectionLostException extends Exception;
  82
  83 class Connection(val in: Reader, val out: Writer)
  84         extends Publisher[AsyncMessage]
  85 {
  86   /* Synchronization.
  87    *
  88    * This class is complicatedly multithreaded.  The following fields must
  89    * only be accessed while the instance is locked.  To prevent deadlocks,
  90    * hold the `Connection' lock before locking any individual `Job' objects.
  91    */
  92
  93   var livep: Boolean = true;            // Is this connection still alive?
  94   var fgjob: Option[this.Job] = None;   // Foreground job, if there is one.
  95   val jobmap = new HashMap[String, this.Job]; // Maps tags to extant jobs.
  96   var bgseq = 0;                        // Next background job tag.
  97
  98   class Job extends Iterator[Seq[String]] {
  99     private[Connection] val ch = new Channel[JobMessage];
 100     private[this] var nextmsg: Option[JobMessage] = None;
 101
 102     private[this] def fetchNext()
 103       { if (nextmsg == None) nextmsg = Some(ch.read); }
 104     override def hasNext: Boolean = {
 105       fetchNext();
 106       nextmsg match {
 107         case Some(JobOK) => false
 108         case _ => true
 109       }
 110     }
 111     override def next(): Seq[String] = {
 112       fetchNext();
 113       nextmsg match {
 114         case None => ???
 115         case Some(JobOK) => throw new NoSuchElementException
 116         case Some(JobFail(msg)) => throw new CommandFailed(msg)
 117         case Some(JobLostConnection) => throw new ConnectionLostException
 118         case Some(JobInfo(msg)) => nextmsg = None; msg
 119       }
 120     }
 121
 122     def keyvals(): Map[String, String] = {
 123       val b = Map.newBuilder[String, String];
 124       for (line <- this; token <- line) {
 125         token.indexOf('=') match {
 126           case -1 => throw new ServerFailed("missing `=' in key-value list");
 127           case eq =>
 128             val k = token.substring(0, eq);
 129             val v = token.substring(eq + 1);
 130             b += k -> v;
 131         }
 132       }
 133       b.result
 134     }
 135
 136     def traceish(): Seq[(Char, Boolean, String)] = {
 137       val b = Seq.newBuilder[(Char, Boolean, String)];
 138       for (line <- this) line match {
 139         case List(key, desc@_*) =>
 140           val live = if (key.length == 1) false
 141                      else if (key.length == 2 && key(1) == '+') true
 142                      else throw new ServerFailed(
 143                        s"incomprehensible traceish key `$key'");
 144           b += ((key(0), live, desc.mkString(" ")));
 145         case _ => throw new ServerFailed("empty line in traceish output");
 146       }
 147       b.result
 148     }
 149
 150     def expectEmpty() {
 151       if (hasNext) throw new ServerFailed("no output expected");
 152     }
 153
 154     def oneLine(): Seq[String] = {
 155       if (hasNext) {
 156         val line = next();
 157         if (!hasNext) return line;
 158       }
 159       throw new ServerFailed("exactly one line expected");
 160     }
 161   }
 162
 163   def submit(bg: Boolean, toks: String*): this.Job = {
 164     var cmd = toks;
 165 println(";; wait for lock");
 166     synchronized {
 167       if (bg) {
 168         val tag = bgseq formatted "J%05d"; bgseq += 1;
 169         cmd = toks match {
 170           case Seq(cmd, tail@_*) => cmd +: "-background" +: tag +: tail;
 171         }
 172       }
 173 println(";; wait for foreground");
 174       while (livep && fgjob != None) wait();
 175       if (!livep) throw new ConnectionClosed;
 176 println(";; write command");
 177       try { out.write(quoteTokens(cmd)); out.write('\n'); out.flush(); }
 178       catch { case e: Throwable => notify(); throw e; }
 179       val j = new Job;
 180       fgjob = Some(j);
 181       j
 182     }
 183   }
 184
 185   def submit(toks: String*): this.Job = submit(false, toks: _*);
 186
 187   def close() { synchronized { out.close(); } }
 188
 189   /* These two expect the connection lock to be held. */
 190   def foregroundJob: Job =
 191     fgjob.getOrElse { throw new ServerFailed("no foreground job"); }
 192   def releaseForegroundJob() { fgjob = None; notify(); }
 193
 194   def parseServerLine(s: String): Message = nextToken(s) match {
 195     case None => throw new ServerFailed("empty line from server")
 196     case Some(("TRACE", next)) => Trace(s.substring(next))
 197     case Some((code, next)) => (code, splitTokens(s, next)) match {
 198       case ("OK", Seq()) => JobOK
 199       case ("INFO", tail) => JobInfo(tail)
 200       case ("FAIL", tail) => JobFail(tail)
 201       case ("BGDETACH", Seq(tag)) => JobDetached(tag)
 202       case ("BGOK", Seq(tag)) => BackgroundJobMessage(tag, JobOK)
 203       case ("BGINFO", Seq(tag, tail@_*)) =>
 204         BackgroundJobMessage(tag, JobInfo(tail))
 205       case ("BGFAIL", Seq(tag, tail@_*)) =>
 206         BackgroundJobMessage(tag, JobFail(tail))
 207       case ("WARN", tail) => Warning(tail)
 208       case ("NOTE", tail) => Notify(tail)
 209       case ("SVCCLAIM", Seq(svc, ver)) => ServiceClaim(svc, ver)
 210       case ("SVCJOB", Seq(tag, svc, cmd, args@_*)) =>
 211         ServiceJob(tag, svc, cmd, args)
 212       case ("SVCCANCEL", Seq(tag)) => ServiceCancel(tag)
 213       case (_, tail) => throw new ServerFailed(
 214         "incomprehensible line from server: " + quoteTokens(code +: tail))
 215     }
 216   }
 217
 218   def processJobMessage(msg: JobMessage)
 219                        (getjob: (Boolean) => Job) {
 220     synchronized { getjob(msg.isInstanceOf[JobInfo]); }.ch.write(msg);
 221   }
 222
 223   /* Reading lines from the server. */
 224   val readthr = thread("admin reader") {
 225 println(";; readthr running");
 226     val bin = in match {
 227       case br: BufferedReader => br;
 228       case _ => new BufferedReader(in)
 229     }
 230     var line: String = null;
 231
 232     try {
 233 println(";; wait for line");
 234       while ({line = bin.readLine; line != null}) {
 235 println(s";; line: $line");
 236         parseServerLine(line) match {
 237           case JobDetached(tag) => synchronized {
 238             jobmap(tag) = foregroundJob; releaseForegroundJob();
 239           }
 240           case msg: JobMessage => processJobMessage(msg) { keep =>
 241             val j = foregroundJob; if (!keep) releaseForegroundJob(); j
 242           }
 243           case BackgroundJobMessage(tag, msg) =>
 244             processJobMessage(msg) { keep =>
 245               val j = jobmap.getOrElse(tag, throw new ServerFailed(
 246                 s"no job with tag `${tag}'"));
 247               if (!keep) jobmap.remove(tag);
 248               j
 249             }
 250           case msg: AsyncMessage =>
 251             publish(msg);
 252           case _: ServiceMessage =>
 253             ();
 254         }
 255       }
 256     } catch {
 257       case e: Throwable => e.printStackTrace();
 258     } finally {
 259       synchronized {
 260         livep = false;
 261         for ((_, j) <- jobmap) j.ch.write(JobLostConnection);
 262         fgjob match {
 263           case Some(j) =>
 264             j.ch.write(JobLostConnection);
 265             fgjob = None;
 266             notifyAll();
 267           case None => ();
 268         }
 269       }
 270       publish(ConnectionLost);
 271       in.close(); out.close();
 272     }
 273   }
 274 }
 275
 276 /*----- That's all, folks -------------------------------------------------*/
 277
 278 }