ref: 3efb5bbb4061056e523858b134c555949591efe2
dir: /appl/cmd/plumber.b/
implement Plumber;
include "sys.m";
sys: Sys;
include "draw.m";
draw: Draw;
include "sh.m";
include "regex.m";
regex: Regex;
include "string.m";
str: String;
include "../lib/plumbing.m";
plumbing: Plumbing;
Pattern, Rule: import plumbing;
include "plumbmsg.m";
plumbmsg: Plumbmsg;
Msg, Attr: import plumbmsg;
include "arg.m";
Plumber: module
{
init: fn(ctxt: ref Draw->Context, argl: list of string);
};
Input: adt
{
inc: chan of ref Inmesg;
resc: chan of int;
io: ref Sys->FileIO;
};
Output: adt
{
name: string;
outc: chan of string;
io: ref Sys->FileIO;
queue: list of array of byte;
started: int;
startup: string;
waiting: int;
};
Port: adt
{
name: string;
startup: string;
alwaysstart: int;
};
Match: adt
{
p0, p1: int;
};
Inmesg: adt
{
msg: ref Msg;
text: string; # if kind is text
p0,p1: int;
match: array of Match;
port: int;
startup: string;
args: list of string;
attrs: list of ref Attr;
clearclick: int;
set: int;
# $ arguments
_n: array of string;
_dir: string;
_file: string;
};
# Message status after processing
HANDLED: con -1;
UNKNOWN: con -2;
NOTSTARTED: con -3;
output: array of ref Output;
input: ref Input;
stderr: ref Sys->FD;
pgrp: int;
rules: list of ref Rule;
titlectl: chan of string;
ports: list of ref Port;
wmstartup := 0;
wmchan := "/chan/wm";
verbose := 0;
context: ref Draw->Context;
usage()
{
sys->fprint(stderr, "Usage: plumb [-vw] [-c wmchan] [initfile ...]\n");
raise "fail:usage";
}
init(ctxt: ref Draw->Context, args: list of string)
{
context = ctxt;
sys = load Sys Sys->PATH;
draw = load Draw Draw->PATH;
stderr = sys->fildes(2);
regex = load Regex Regex->PATH;
plumbing = load Plumbing Plumbing->PATH;
str = load String String->PATH;
err: string;
nogrp := 0;
arg := load Arg Arg->PATH;
arg->init(args);
while ((opt := arg->opt()) != 0) {
case opt {
'w' =>
wmstartup = 1;
'c' =>
if ((wmchan = arg->arg()) == nil)
usage();
'v' =>
verbose = 1;
'n' =>
nogrp = 1;
* =>
usage();
}
}
args = arg->argv();
arg = nil;
(rules, err) = plumbing->init(regex, args);
if(err != nil){
sys->fprint(stderr, "plumb: %s\n", err);
raise "fail:init";
}
plumbmsg = load Plumbmsg Plumbmsg->PATH;
plumbmsg->init(0, nil, 0);
if(nogrp)
pgrp = sys->pctl(0, nil);
else
pgrp = sys->pctl(sys->NEWPGRP, nil);
r := rules;
for(i:=0; i<len rules; i++){
rule := hd r;
r = tl r;
for(j:=0; j<len rule.action; j++)
if(rule.action[j].pred == "to" || rule.action[j].pred == "alwaysstart"){
p := findport(rule.action[j].arg);
if(p == nil){
p = ref Port(rule.action[j].arg, nil, rule.action[j].pred == "alwaysstart");
ports = p :: ports;
}
for(k:=0; k<len rule.action; k++)
if(rule.action[k].pred == "start")
p.startup = rule.action[k].arg;
break;
}
}
input = ref Input;
input.io = makefile("plumb.input");
if(input.io == nil)
shutdown();
input.inc = chan of ref Inmesg;
input.resc = chan of int;
spawn receiver(input);
output = array[len ports] of ref Output;
pp := ports;
for(i=0; i<len output; i++){
p := hd pp;
pp = tl pp;
output[i] = ref Output;
output[i].name = p.name;
output[i].io = makefile("plumb."+p.name);
if(output[i].io == nil)
shutdown();
output[i].outc = chan of string;
output[i].started = 0;
output[i].startup = p.startup;
output[i].waiting = 0;
}
# spawn so we return without needing to run plumb in background
spawn sender(input, output);
}
findport(name: string): ref Port
{
for(p:=ports; p!=nil; p=tl p)
if((hd p).name == name)
return hd p;
return nil;
}
makefile(file: string): ref Sys->FileIO
{
io := sys->file2chan("/chan", file);
if(io == nil){
sys->fprint(stderr, "plumb: can't establish /chan/%s: %r\n", file);
return nil;
}
return io;
}
receiver(input: ref Input)
{
for(;;){
(nil, msg, nil, wc) := <-input.io.write;
if(wc == nil)
; # not interested in EOF; leave channel open
else{
input.inc <-= parse(msg);
res := <- input.resc;
err := "";
if(res == UNKNOWN)
err = "no matching plumb rule";
wc <-= (len msg, err);
}
}
}
sender(input: ref Input, output: array of ref Output)
{
outputc := array[len output] of chan of (int, int, int, Sys->Rread);
for(;;){
alt{
in := <-input.inc =>
if(in == nil){
input.resc <-= HANDLED;
break;
}
(j, msg) := process(in);
case j {
HANDLED =>
break;
UNKNOWN =>
if(in.msg.src != "acme")
sys->fprint(stderr, "plumb: don't know who message goes to\n");
NOTSTARTED =>
sys->fprint(stderr, "plumb: can't start application\n");
* =>
output[j].queue = append(output[j].queue, msg);
outputc[j] = output[j].io.read;
}
input.resc <-= j;
(j, tmp) := <-outputc =>
(nil, nbytes, nil, rc) := tmp;
if(rc == nil) # no interest in EOF
break;
msg := hd output[j].queue;
if(nbytes < len msg){
rc <-= (nil, "buffer too short for message");
break;
}
output[j].queue = tl output[j].queue;
if(output[j].queue == nil)
outputc[j] = nil;
rc <-= (msg, nil);
}
}
}
parse(a: array of byte): ref Inmesg
{
msg := Msg.unpack(a);
if(msg == nil)
return nil;
i := ref Inmesg;
i.msg = msg;
if(msg.dst != nil){
if(control(i))
return nil;
toport(i, msg.dst);
}else
i.port = -1;
i.match = array[10] of { * => Match(-1, -1)};
i._n = array[10] of string;
i.attrs = plumbmsg->string2attrs(i.msg.attr);
return i;
}
append(l: list of array of byte, a: array of byte): list of array of byte
{
if(l == nil)
return a :: nil;
return hd l :: append(tl l, a);
}
shutdown()
{
fname := sys->sprint("#p/%d/ctl", pgrp);
if((fdesc := sys->open(fname, sys->OWRITE)) != nil)
sys->write(fdesc, array of byte "killgrp\n", 8);
raise "fail:error";
}
# Handle control messages
control(in: ref Inmesg): int
{
msg := in.msg;
if(msg.kind!="text" || msg.dst!="plumb")
return 0;
text := string msg.data;
case text {
"start" =>
start(msg.src, 1);
"stop" =>
start(msg.src, -1);
* =>
sys->fprint(stderr, "plumb: unrecognized control message from %s: %s\n", msg.src, text);
}
return 1;
}
start(port: string, startstop: int)
{
for(i:=0; i<len output; i++)
if(port == output[i].name){
output[i].waiting = 0;
output[i].started += startstop;
return;
}
sys->fprint(stderr, "plumb: \"start\" message from unrecognized port %s\n", port);
}
startup(dir, prog: string, args: list of string, wait: chan of int)
{
if(wmstartup){
fd := sys->open(wmchan, Sys->OWRITE);
if(fd != nil){
sys->fprint(fd, "s %s", str->quoted(dir :: prog :: args));
wait <-= 1;
return;
}
}
sys->pctl(Sys->NEWFD|Sys->NEWPGRP|Sys->FORKNS, list of {0, 1, 2});
wait <-= 1;
wait = nil;
mod := load Command prog;
if(mod == nil){
sys->fprint(stderr, "plumb: can't load %s: %r\n", prog);
return;
}
sys->chdir(dir);
mod->init(context, prog :: args);
}
# See if messages should be queued while waiting for program to connect
shouldqueue(out: ref Output): int
{
p := findport(out.name);
if(p == nil){
sys->fprint(stderr, "plumb: can't happen in shouldqueue\n");
return 0;
}
if(p.alwaysstart)
return 0;
return out.waiting;
}
# Determine destination of input message, reformat for output
process(in: ref Inmesg): (int, array of byte)
{
if(!clarify(in))
return (UNKNOWN, nil);
if(in.port < 0)
return (UNKNOWN, nil);
a := in.msg.pack();
j := in.port;
if(a == nil)
j = UNKNOWN;
else if(output[j].started==0 && !shouldqueue(output[j])){
path: string;
args: list of string;
if(in.startup!=nil){
path = macro(in, in.startup);
args = expand(in, in.args);
}else if(output[j].startup != nil){
path = output[j].startup;
args = in.text :: nil;
}else
return (NOTSTARTED, nil);
log(sys->sprint("start %s port %s\n", path, output[j].name));
wait := chan of int;
output[j].waiting = 1;
spawn startup(in.msg.dir, path, args, wait);
<-wait;
return (HANDLED, nil);
}else{
if(in.msg.kind != "text")
text := sys->sprint("message of type %s", in.msg.kind);
else{
text = in.text;
for(i:=0; i<len text; i++){
if(text[i]=='\n'){
text = text[0:i];
break;
}
if(i > 50) {
text = text[0:i]+"...";
break;
}
}
}
log(sys->sprint("send \"%s\" to %s", text, output[j].name));
}
return (j, a);
}
# expand $arguments
expand(in: ref Inmesg, args: list of string): list of string
{
a: list of string;
while(args != nil){
a = macro(in, hd args) :: a;
args = tl args;
}
while(a != nil){
args = hd a :: args;
a = tl a;
}
return args;
}
# resolve all ambiguities, fill in any missing fields
clarify(in: ref Inmesg): int
{
in.clearclick = 0;
in.set = 0;
msg := in.msg;
if(msg.kind != "text")
return 0;
in.text = string msg.data;
if(msg.dst != "")
return 1;
return dorules(in, rules);
}
dorules(in: ref Inmesg, rules: list of ref Rule): int
{
if (verbose)
log("msg: " + inmesg2s(in));
for(r:=rules; r!=nil; r=tl r) {
if(matchrule(in, hd r)){
applyrule(in, hd r);
if (verbose)
log("yes");
return 1;
} else if (verbose)
log("no");
}
return 0;
}
inmesg2s(in: ref Inmesg): string
{
m := in.msg;
s := sys->sprint("src=%s; dst=%s; dir=%s; kind=%s; attr='%s'",
m.src, m.dst, m.dir, m.kind, m.attr);
if (m.kind == "text")
s += "; data='" + string m.data + "'";
return s;
}
matchrule(in: ref Inmesg, r: ref Rule): int
{
pats := r.pattern;
for(i:=0; i<len in.match; i++)
in.match[i] = (-1,-1);
# no rules at all implies success, so return if any fail
for(i=0; i<len pats; i++)
if(matchpattern(in, pats[i]) == 0)
return 0;
return 1;
}
applyrule(in: ref Inmesg, r: ref Rule)
{
acts := r.action;
for(i:=0; i<len acts; i++)
applypattern(in, acts[i]);
if(in.clearclick){
al: list of ref Attr;
for(l:=in.attrs; l!=nil; l=tl l)
if((hd l).name != "click")
al = hd l :: al;
in.attrs = al;
in.msg.attr = plumbmsg->attrs2string(al);
if(in.set){
in.text = macro(in, "$0");
in.msg.data = array of byte in.text;
}
}
}
matchpattern(in: ref Inmesg, p: ref Pattern): int
{
msg := in.msg;
text: string;
case p.field {
"src" => text = msg.src;
"dst" => text = msg.dst;
"dir" => text = msg.dir;
"kind" => text = msg.kind;
"attr" => text = msg.attr;
"data" => text = in.text;
* =>
sys->fprint(stderr, "plumb: don't recognize pattern field %s\n", p.field);
return 0;
}
if (verbose)
log(sys->sprint("'%s' %s '%s'\n", text, p.pred, p.arg));
case p.pred {
"is" =>
return text == p.arg;
"isfile" or "isdir" =>
text = p.arg;
if(p.expand)
text = macro(in, text);
if(len text == 0)
return 0;
if(len in.msg.dir!=0 && text[0] != '/' && text[0]!='#')
text = in.msg.dir+"/"+text;
text = cleanname(text);
(ok, dir) := sys->stat(text);
if(ok < 0)
return 0;
if(p.pred=="isfile" && (dir.mode&Sys->DMDIR)==0){
in._file = text;
return 1;
}
if(p.pred=="isdir" && (dir.mode&Sys->DMDIR)!=0){
in._dir = text;
return 1;
}
return 0;
"matches" =>
(clickspecified, val) := plumbmsg->lookup(in.attrs, "click");
if(p.field != "data")
clickspecified = 0;
if(!clickspecified){
# easy case. must match whole string
matches := regex->execute(p.regex, text);
if(matches == nil)
return 0;
(p0, p1) := matches[0];
if(p0!=0 || p1!=len text)
return 0;
in.match = matches;
setvars(in, text);
return 1;
}
matches := clickmatch(p.regex, text, int val);
if(matches == nil)
return 0;
(p0, p1) := matches[0];
# assumes all matches are in same sequence
if(in.match[0].p0 != -1)
return p0==in.match[0].p0 && p1==in.match[0].p1;
in.match = matches;
setvars(in, text);
in.clearclick = 1;
in.set = 1;
return 1;
"set" =>
text = p.arg;
if(p.expand)
text = macro(in, text);
case p.field {
"src" => msg.src = text;
"dst" => msg.dst = text;
"dir" => msg.dir = text;
"kind" => msg.kind = text;
"attr" => msg.attr = text;
"data" => in.text = text;
msg.data = array of byte text;
msg.kind = "text";
in.set = 0;
}
return 1;
* =>
sys->fprint(stderr, "plumb: don't recognize pattern predicate %s\n", p.pred);
}
return 0;
}
applypattern(in: ref Inmesg, p: ref Pattern): int
{
if(p.field != "plumb"){
sys->fprint(stderr, "plumb: don't recognize action field %s\n", p.field);
return 0;
}
case p.pred {
"to" or "alwaysstart" =>
if(in.port >= 0) # already specified
return 1;
toport(in, p.arg);
"start" =>
in.startup = p.arg;
in.args = p.extra;
* =>
sys->fprint(stderr, "plumb: don't recognize action %s\n", p.pred);
}
return 1;
}
toport(in: ref Inmesg, name: string): int
{
for(i:=0; i<len output; i++)
if(name == output[i].name){
in.msg.dst = name;
in.port = i;
return i;
}
in.port = -1;
sys->fprint(stderr, "plumb: unrecognized port %s\n", name);
return -1;
}
# simple heuristic: look for leftmost match that reaches click position
clickmatch(re: ref Regex->Arena, text: string, click: int): array of Match
{
for(i:=0; i<=click && i < len text; i++){
matches := regex->executese(re, text, (i, -1), i == 0, 1);
if(matches == nil)
continue;
(p0, p1) := matches[0];
if(p0>=i && p1>=click)
return matches;
}
return nil;
}
setvars(in: ref Inmesg, text: string)
{
for(i:=0; i<len in.match && in.match[i].p0>=0; i++)
in._n[i] = text[in.match[i].p0:in.match[i].p1];
for(; i<len in._n; i++)
in._n[i] = "";
}
macro(in: ref Inmesg, text: string): string
{
word := "";
i := 0;
j := 0;
for(;;){
if(i == len text)
break;
if(text[i++] != '$')
continue;
if(i == len text)
break;
word += text[j:i-1];
(res, skip) := dollar(in, text[i:]);
word += res;
i += skip;
j = i;
}
if(j < len text)
word += text[j:];
return word;
}
dollar(in: ref Inmesg, text: string): (string, int)
{
if(text[0] == '$')
return ("$", 1);
if('0'<=text[0] && text[0]<='9')
return (in._n[text[0]-'0'], 1);
if(len text < 3)
return ("$", 0);
case text[0:3] {
"src" => return (in.msg.src, 3);
"dst" => return (in.msg.dst, 3);
"dir" => return (in._dir, 3);
}
if(len text< 4)
return ("$", 0);
case text[0:4] {
"attr" => return (in.msg.attr, 4);
"data" => return (in.text, 4);
"file" => return (in._file, 4);
"kind" => return (in.msg.kind, 4);
}
return ("$", 0);
}
# compress ../ references and do other cleanups
cleanname(name: string): string
{
# compress multiple slashes
n := len name;
for(i:=0; i<n-1; i++)
if(name[i]=='/' && name[i+1]=='/'){
name = name[0:i]+name[i+1:];
--i;
n--;
}
# eliminate ./
for(i=0; i<n-1; i++)
if(name[i]=='.' && name[i+1]=='/' && (i==0 || name[i-1]=='/')){
name = name[0:i]+name[i+2:];
--i;
n -= 2;
}
found: int;
do{
# compress xx/..
found = 0;
for(i=1; i<=n-3; i++)
if(name[i:i+3] == "/.."){
if(i==n-3 || name[i+3]=='/'){
found = 1;
break;
}
}
if(found)
for(j:=i-1; j>=0; --j)
if(j==0 || name[j-1]=='/'){
i += 3; # character beyond ..
if(i<n && name[i]=='/')
++i;
name = name[0:j]+name[i:];
n -= (i-j);
break;
}
}while(found);
# eliminate trailing .
if(n>=2 && name[n-2]=='/' && name[n-1]=='.')
--n;
if(n == 0)
return ".";
if(n != len name)
name = name[0:n];
return name;
}
log(s: string)
{
if(len s == 0)
return;
if(s[len s-1] != '\n')
s[len s] = '\n';
sys->print("plumb: %s", s);
}