path: root/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);
}