path: root/appl/svc/webget/webget.b

implement Webget;

# Protocol
#
# Client opens /chan/webget and writes one of
#		GET  0 reqid url types cachectl authcookie\n
#	    or
#		POST bodylength reqid url types cachectl authcookie\n
#		body
#
# The possibilities for cachectl are
#		max-stale=seconds
#			client is willing to accept a response whose age exceeds
#			its freshness lifetime (by at most specified seconds)
#			without revalidation
#		max-age=seconds
#			client is unwilling to accept a response whose age
#			(now - generation time) exceeds specified seconds
#			without revalidiation
#		no-cache
#			unconditional reload
# Both max-stale and max-age may be specified (separated by comma),
# but no-cache must appear by itself.
#
# Authcookie is optional.  If present, it goes in an Authorization: header.
#
# The appropriate transport mechanism gets the entity and
# responds with one of
#		OK bodylength reqid type url\n
#		body
#	    or
#		ERROR reqid message\n
#
# (In the ERROR case, the message might be "Unauthorized: challenge\n",
# where challenge is of the form "BASIC realm=xxx (param, param, ...)\n".
# The user can be prompted for a name:password, and the request repeated
# with authcookie containing the base64 encoding of name:password).

include	"sys.m";
	sys: Sys;
	FD: import sys;

include "draw.m";

include "string.m";
	S: String;

include "bufio.m";
	B: Bufio;

include "dial.m";
	DI: Dial;

include "message.m";
	M: Message;
	Msg: import M;

include "url.m";
	U: Url;
	ParsedUrl: import U;

include "webget.m";

include "wgutils.m";
	W: WebgetUtils;
	Fid, Req: import W;

include "transport.m";
	
fhash := array[128] of ref Fid;

Transports: adt
{
	scheme:		int;
	m:		Transport;
};
transports: array of ref Transports;

transtab := array[] of {
	(Url->HTTP,	"/dis/svc/webget/http.dis"),
	(Url->HTTPS,	nil),	# nil means: same as previous
	(Url->FILE,	"/dis/svc/webget/file.dis"),
	(Url->FTP,	"/dis/svc/webget/ftp.dis")
};

Transpreq: adt
{
	index: int;
	fid: ref Fid;
	r: ref Req;
	next: cyclic ref Transpreq;
};

Rchunk: con 30;
# Transpmax: con 5;	# max number of simultaneously spawned transports
Transpmax: con 1;	# max number of simultaneously spawned transports

logfile: con "/services/webget/webget.log";
DO_LOG: con 1;

stderr: ref FD;

# to start ever-present webget
init(nil: ref Draw->Context, nil: list of string)
{
	dummyctl := chan of int;
	spawn start(dummyctl);
	<- dummyctl;
	<- dummyctl;
}

# sends a 1 on ctl when ready to serve,
# 0 if there was some problem starting.
start(ctl: chan of int)
{
	sys = load Sys Sys->PATH;
	stderr = sys->fildes(2);
	ok := 1;
	ntransp := 0;
	tqueuehd: ref Transpreq = nil;
	tqueuetl: ref Transpreq = nil;

	log : ref Sys->FD;
	if(DO_LOG)
		log = sys->create(logfile, Sys->OWRITE, 8r666);

	io := sys->file2chan("/chan", "webget");
	if(io == nil) {
		sys->fprint(stderr, "webget: failed to post: %r\n");
		ok = 0;
	}

	B = load Bufio Bufio->PATH;
	if(B == nil) {
		sys->fprint(stderr, "webget: failed to load Bufio: %r\n");
		ok = 0;
	}
	S = load String String->PATH;
	if(S == nil) {
		sys->fprint(stderr, "webget: failed to load String: %r\n");
		ok = 0;
	}
	M = load Message Message->PATH;
	if(M == nil) {
		sys->fprint(stderr, "webget: failed to load Message: %r\n");
		ok = 0;
	}
	M->init(B, S);
	U = load Url Url->PATH;
	if(U == nil) {
		sys->fprint(stderr, "webget: failed to load Url: %r\n");
		ok = 0;
	}
	U->init();
	W = load WebgetUtils WebgetUtils->PATH;
	if(W == nil) {
		sys->fprint(stderr, "webget: failed to load WebgetUtils: %r\n");
		ok = 0;
	}
	if(!ok) {
		ctl <-= 0;
		return;
	}
	W->init(M, S, B, U, DI, log);

	loadtransmod();

	donec := chan of ref Fid;
	ctl <-= 1;
	

    altloop:
	for(;;) alt {
	(nil, data, fid, wc) := <-io.write =>
		if(wc == nil) {
			finish(fid);
			continue altloop;
		}
		ndata := len data;
		c := lookup(fid);
		W->log(c, "\nREQUEST: " + string data);
		iw := c.writer;
		n := len c.reqs;
		if(iw == n) {
			newrs := array[n + Rchunk] of ref Req;
			newrs[0:] = c.reqs[0:n];
			c.reqs = newrs;
		}
		r := c.reqs[iw];
		err := "";
		if(r == nil) {
			# initial part (or all) of a request
			r = ref Req(iw, "", 0, "", "", "", "", "", nil, nil, nil);
			c.reqs[iw] = r;

			# expect at least the prefix line to be in data
			prefix := "";
			for(i := 0; i < ndata; i++) {
				if(int data[i] == '\n') {
					prefix = string data[0:i];
					if(i+1 < ndata) {
						r.body = array[ndata-i-1] of byte;
						r.body[0:] = data[i+1:ndata];
					}
					break;
				}
			}
			if(prefix == "")
				err = "no prefix line";
			else if(prefix == "FINISH") {
				writereply(wc, len data, "");
				finish(fid);
				continue altloop;
			}
			else {
				(nl, l) := sys->tokenize(prefix, "∎");
				if(nl != 6 && nl != 7)
					err = "wrong number of fields in " + prefix;
				else {
					r.method = hd l;
					l = tl l;
					r.bodylen = int hd(l);
					l = tl l;
					r.reqid = hd l;
					l = tl l;
					r.loc = hd l;
					l = tl l;
					r.types = hd l;
					l = tl l;
					r.cachectl = hd l;
					l = tl l;
					if(l != nil)
						r.auth = hd l;
					locurl := U->makeurl(r.loc);
					if(locurl.scheme == U->MAILTO)
						err = "webget shouldn't get mailto";
					else if(locurl.scheme == U->NOSCHEME || 
					   (locurl.scheme != U->FILE && (locurl.host == "" || locurl.pstart != "/")))
						err = "url not absolute: " + r.loc;
					r.url = locurl;
				}
			}
			if(err != "")
				err = "webget protocol error: " + err;
		}
		else {
			# continuation of request: more body
			olen := len r.body;
			newa := array[olen + ndata] of byte;
			newa[0:] = r.body[0:olen];
			newa[olen:] = data[0:ndata];
			r.body = newa;
		}
		if(err == "" && len r.body == r.bodylen) {
			# request complete: spawn off transport handler
			c.writer++;
			scheme := r.url.scheme;
			found := 0;
			for(i := 0; i < len transports; i++) {
				if(transports[i].scheme == scheme) {
					found = 1;
					break;
				}
			}
			if(found == 0)
				err = "don't know how to fetch " + r.loc;
			else {
				if(ntransp < Transpmax) {
					W->log(c, "transport " + string scheme + ":  get " + r.loc);
					spawn transports[i].m->connect(c, r, donec);
					ntransp++;
				}
				else {
					# too many active transports: queue this until later
					tr := ref Transpreq(i, c, r, nil);
					if(tqueuetl == nil)
						tqueuehd = tqueuetl = tr;
					else {
						tqueuetl.next = tr;
						tqueuetl = tr;
					}
				}
			}
		}
		if(err != "") {
			writereply(wc, -1, err);
			W->log(c, err);
			c.reqs[iw] = nil;
		}
		else
			writereply(wc, ndata, "");

	(nil, nbyte, fid, rc) := <-io.read =>
		if(rc == nil) {
			finish(fid);
			continue altloop;
		}
		c := lookup(fid);
		c.nbyte = nbyte;
		c.rc = rc;
		readans(c);
	c := <- donec =>
		ntransp--;
		if(tqueuehd != nil) {
			tr := tqueuehd;
			tqueuehd = tr.next;
			if(tqueuehd == nil)
				tqueuetl = nil;
			W->log(c, "transport:  get " + tr.r.loc);
			spawn transports[tr.index].m->connect(tr.fid, tr.r, donec);
			ntransp++;
		}
		readans(c);
		c = nil;
	}
}

loadtransmod()
{
	transports = array[len transtab] of ref Transports;
	j := 0;
	prevt : ref Transports = nil;
	for(i := 0; i < len transtab; i++) {
		(scheme, path) := transtab[i];
		if(path == nil) {
			if(prevt != nil)
				transports[j++] = ref Transports(scheme, prevt.m);
		}
		else {
			t := load Transport path;
			if(t == nil) {
				sys->fprint(stderr, "failed to load %s: %r\n", path);
				continue;
			}
	
			t->init(W);

			ts := ref Transports(scheme, t);
			transports[j++] = ts;
			prevt = ts;
		}
	}
}

# Answer a read request c.nbyte bytes, reply to go to c.rc.
# If c.readr is not -1, it is the index of a req with the currently
# being consumed reply.
# c.nread contains the number of bytes of this message read so far.
readans(c: ref Fid)
{
	n := c.nbyte;
	if(n <= 0)
		return;
	ir := c.readr;
	if(ir == -1) {
		# look for ready reply
		for(i := 0; i < c.writer; i++) {
			r := c.reqs[i];
			if(r != nil && r.reply != nil)
				break;
		}
		if(i == c.writer) {
			return;
		}
		ir = i;
	}
	r := c.reqs[ir];
	m := r.reply;
	if(m == nil) {
		W->log(c, "readans bad state: nil reply");
		readreply(c, nil, "");
		return;
	}
	if(m.prefixbytes == nil && m.prefixline != "")
		m.prefixbytes = array of byte m.prefixline;
	plen := len m.prefixbytes;
	blen := m.bodylen;	
	ntot := plen + blen;
	nread := c.nread;
	if(nread == 0)
		W->log(c, "\nREPLY: " + m.prefixline);
	nrest := ntot - nread;
	if(nrest <= 0) {
		W->log(c, "readans bad state: 0 left");
		readreply(c, nil, "");
		return;
	}
	if(n > nrest)
		n = nrest;
	n1 := plen - nread;
	if(n1 > 0) {
		if(n1 > n)
			n1 = n;
		readreply(c, m.prefixbytes[nread:nread + n1], "");
		c.nread += n1;
	}
	else {
		bpos := nread - plen;
		n2 := blen - bpos;
		if(n > n2)
			n = n2;
		readreply(c, m.body[bpos:bpos+n], "");
		c.nread += n;
	}
	if(c.nread >= ntot) {
		c.reqs[ir] = nil;
		c.readr = -1;
		c.nbyte = 0;
		c.nread = 0;
		c.rc = nil;
		# move back write pointer as far as possible
		if(c.writer == ir+1) {
			while(ir >= 0 && c.reqs[ir] == nil)
				ir--;
			c.writer = ir+1;
		}
	}
	else
		c.readr = ir;
}

# Reply to a write command.
writereply(wc: Sys->Rwrite, n: int, err: string)
{
	wc <-= (n, err);
}

readreply(c: ref Fid, a: array of byte, err: string)
{
	rc := c.rc;
	if(rc != nil)
		rc <-= (a, err);
	c.nbyte = 0;
}

lookup(fid: int): ref Fid
{
	h := fid%len fhash;
	for(f := fhash[h]; f != nil; f = f.link)
		if(f.fid == fid)
			return f;
	f = ref Fid(fid, fhash[h], array[Rchunk] of ref Req, 0, -1, 0, 0, nil);
	fhash[h] = f;

	W->log(f, "\nNEW CLIENT");

	return f;	
}

finish(fid: int)
{
	W->log(nil, "finish");
	h := fid%len fhash;

	prev: ref Fid;
	for(f := fhash[h]; f != nil; f = f.link) {
		if(f.fid == fid) {
			f.rc = nil;
			W->log(f, "client finished");
			if(prev == nil)
				fhash[h] = f.link;
			else
				prev.link = f.link;
			return;
		}
	}
}