Single-threaded plumber that can run "start" rules.

Thanks to Caerwyn Jones.

3 files changed, 530 insertions, 0 deletions

diff --git a/src/libplumb/Makefile b/src/libplumb/Makefile
new file mode 100755
index 00000000..e9d07a66
--- /dev/null
+++ b/src/libplumb/Makefile
@@ -0,0 +1,19 @@
+
+LIB=../../lib/libplumb.a
+OFILES=\
+	mesg.o\
+
+HFILES=../../include/plumb.h
+
+INCLUDES=-I../../include
+
+CFLAGS += $(INCLUDES) -D_POSIX_SOURCE
+
+CC=cc
+
+$(LIB): $(OFILES)
+	ar r $(LIB) $(OFILES)
+
+clean:
+	rm -rf $(TARG) $(OFILES)
+
diff --git a/src/libplumb/event.c b/src/libplumb/event.c
new file mode 100755
index 00000000..36a95d68
--- /dev/null
+++ b/src/libplumb/event.c
@@ -0,0 +1,108 @@
+#include <u.h>
+#include <libc.h>
+#include <draw.h>
+#include <event.h>
+#include "plumb.h"
+
+typedef struct EQueue EQueue;
+
+struct EQueue
+{
+	int		id;
+	char		*buf;
+	int		nbuf;
+	EQueue	*next;
+};
+
+static	EQueue	*equeue;
+static	Lock		eqlock;
+
+static
+int
+partial(int id, Event *e, uchar *b, int n)
+{
+	EQueue *eq, *p;
+	int nmore;
+
+	lock(&eqlock);
+	for(eq = equeue; eq != nil; eq = eq->next)
+		if(eq->id == id)
+			break;
+	unlock(&eqlock);
+	if(eq == nil)
+		return 0;
+	/* partial message exists for this id */
+	eq->buf = realloc(eq->buf, eq->nbuf+n);
+	if(eq->buf == nil)
+		drawerror(display, "eplumb: cannot allocate buffer");
+	memmove(eq->buf+eq->nbuf, b, n);
+	eq->nbuf += n;
+	e->v = plumbunpackpartial((char*)eq->buf, eq->nbuf, &nmore);
+	if(nmore == 0){	/* no more to read in this message */
+		lock(&eqlock);
+		if(eq == equeue)
+			equeue = eq->next;
+		else{
+			for(p = equeue; p!=nil && p->next!=eq; p = p->next)
+				;
+			if(p == nil)
+				drawerror(display, "eplumb: bad event queue");
+			p->next = eq->next;
+		}
+		unlock(&eqlock);
+		free(eq->buf);
+		free(eq);
+	}
+	return 1;
+}
+
+static
+void
+addpartial(int id, char *b, int n)
+{
+	EQueue *eq;
+
+	eq = malloc(sizeof(EQueue));
+	if(eq == nil)
+		return;
+	eq->id = id;
+	eq->nbuf = n;
+	eq->buf = malloc(n);
+	if(eq->buf == nil){
+		free(eq);
+		return;
+	}
+	memmove(eq->buf, b, n);
+	lock(&eqlock);
+	eq->next = equeue;
+	equeue = eq;
+	unlock(&eqlock);
+}
+
+static
+int
+plumbevent(int id, Event *e, uchar *b, int n)
+{
+	int nmore;
+
+	if(partial(id, e, b, n) == 0){
+		/* no partial message already waiting for this id */
+		e->v = plumbunpackpartial((char*)b, n, &nmore);
+		if(nmore > 0)	/* incomplete message */
+			addpartial(id, (char*)b, n);
+	}
+	if(e->v == nil)
+		return 0;
+	return id;
+}
+
+int
+eplumb(int key, char *port)
+{
+	int fd;
+
+	fd = plumbopen(port, OREAD|OCEXEC);
+	if(fd < 0)
+		return -1;
+	return estartfn(key, fd, 8192, plumbevent);
+}
diff --git a/src/libplumb/mesg.c b/src/libplumb/mesg.c
new file mode 100755
index 00000000..93ab03f6
--- /dev/null
+++ b/src/libplumb/mesg.c
@@ -0,0 +1,403 @@
+#include <u.h>
+#include <libc.h>
+#include "plumb.h"
+
+static char attrbuf[4096];
+
+int
+plumbopen(char *name, int omode)
+{
+	int fd, f;
+	char *s;
+	char buf[128];
+
+	if(name[0] == '/')
+		return open(name, omode);
+	snprint(buf, sizeof buf, "/mnt/plumb/%s", name);
+	fd = open(buf, omode);
+	if(fd >= 0)
+		return fd;
+	snprint(buf, sizeof buf, "/mnt/term/mnt/plumb/%s", name);
+	fd = open(buf, omode);
+	if(fd >= 0)
+		return fd;
+	/* try mounting service */
+	s = getenv("plumbsrv");
+	if(s == nil)
+		return -1;
+	snprint(buf, sizeof buf, "/mnt/plumb/%s", name);
+	return open(buf, omode);
+}
+
+static int
+Strlen(char *s)
+{
+	if(s == nil)
+		return 0;
+	return strlen(s);
+}
+
+static char*
+Strcpy(char *s, char *t)
+{
+	if(t == nil)
+		return s;
+	return strcpy(s, t) + strlen(t);
+}
+
+/* quote attribute value, if necessary */
+static char*
+quote(char *s)
+{
+	char *t;
+	int c;
+
+	if(s == nil){
+		attrbuf[0] = '\0';
+		return attrbuf;
+	}
+	if(strpbrk(s, " '=\t") == nil)
+		return s;
+	t = attrbuf;
+	*t++ = '\'';
+	while(t < attrbuf+sizeof attrbuf-2){
+		c = *s++;
+		if(c == '\0')
+			break;
+		*t++ = c;
+		if(c == '\'')
+			*t++ = c;
+	}
+	*t++ = '\'';
+	*t = '\0';
+	return attrbuf;
+}
+
+char*
+plumbpackattr(Plumbattr *attr)
+{
+	int n;
+	Plumbattr *a;
+	char *s, *t;
+
+	if(attr == nil)
+		return nil;
+	n = 0;
+	for(a=attr; a!=nil; a=a->next)
+		n += Strlen(a->name) + 1 + Strlen(quote(a->value)) + 1;
+	s = malloc(n);
+	if(s == nil)
+		return nil;
+	t = s;
+	*t = '\0';
+	for(a=attr; a!=nil; a=a->next){
+		if(t != s)
+			*t++ = ' ';
+		strcpy(t, a->name);
+		strcat(t, "=");
+		strcat(t, quote(a->value));
+		t += strlen(t);
+	}
+	if(t > s+n)
+		abort();
+	return s;
+}
+
+char*
+plumblookup(Plumbattr *attr, char *name)
+{
+	while(attr){
+		if(strcmp(attr->name, name) == 0)
+			return attr->value;
+		attr = attr->next;
+	}
+	return nil;
+}
+
+char*
+plumbpack(Plumbmsg *m, int *np)
+{
+	int n, ndata;
+	char *buf, *p, *attr;
+
+	ndata = m->ndata;
+	if(ndata < 0)
+		ndata = Strlen(m->data);
+	attr = plumbpackattr(m->attr);
+	n = Strlen(m->src)+1 + Strlen(m->dst)+1 + Strlen(m->wdir)+1 +
+		Strlen(m->type)+1 + Strlen(attr)+1 + 16 + ndata;
+	buf = malloc(n+1);	/* +1 for '\0' */
+	if(buf == nil){
+		free(attr);
+		return nil;
+	}
+	p = Strcpy(buf, m->src);
+	*p++ = '\n';
+	p = Strcpy(p, m->dst);
+	*p++ = '\n';
+	p = Strcpy(p, m->wdir);
+	*p++ = '\n';
+	p = Strcpy(p, m->type);
+	*p++ = '\n';
+	p = Strcpy(p, attr);
+	*p++ = '\n';
+	p += sprint(p, "%d\n", ndata);
+	memmove(p, m->data, ndata);
+	*np = (p-buf)+ndata;
+	buf[*np] = '\0';	/* null terminate just in case */
+	if(*np >= n+1)
+		abort();
+	free(attr);
+	return buf;
+}
+
+int
+plumbsend(int fd, Plumbmsg *m)
+{
+	char *buf;
+	int n;
+
+	buf = plumbpack(m, &n);
+	if(buf == nil)
+		return -1;
+	n = write(fd, buf, n);
+	free(buf);
+	return n;
+}
+
+static int
+plumbline(char **linep, char *buf, int i, int n, int *bad)
+{
+	int starti;
+	char *p;
+
+	if(*bad)
+		return i;
+	starti = i;
+	while(i<n && buf[i]!='\n')
+		i++;
+	if(i == n)
+		*bad = 1;
+	else{
+		p = malloc((i-starti) + 1);
+		if(p == nil)
+			*bad = 1;
+		else{
+			memmove(p, buf+starti, i-starti);
+			p[i-starti] = '\0';
+		}
+		*linep = p;
+		i++;
+	}
+	return i;
+}
+
+void
+plumbfree(Plumbmsg *m)
+{
+	Plumbattr *a, *next;
+
+	free(m->src);
+	free(m->dst);
+	free(m->wdir);
+	free(m->type);
+	for(a=m->attr; a!=nil; a=next){
+		next = a->next;
+		free(a->name);
+		free(a->value);
+		free(a);
+	}
+	free(m->data);
+	free(m);
+}
+
+Plumbattr*
+plumbunpackattr(char *p)
+{
+	Plumbattr *attr, *prev, *a;
+	char *q, *v;
+	int c, quoting;
+
+	attr = prev = nil;
+	while(*p!='\0' && *p!='\n'){
+		while(*p==' ' || *p=='\t')
+			p++;
+		if(*p == '\0')
+			break;
+		for(q=p; *q!='\0' && *q!='\n' && *q!=' ' && *q!='\t'; q++)
+			if(*q == '=')
+				break;
+		if(*q != '=')
+			break;	/* malformed attribute */
+		a = malloc(sizeof(Plumbattr));
+		if(a == nil)
+			break;
+		a->name = malloc(q-p+1);
+		if(a->name == nil){
+			free(a);
+			break;
+		}
+		memmove(a->name, p, q-p);
+		a->name[q-p] = '\0';
+		/* process quotes in value */
+		q++;	/* skip '=' */
+		v = attrbuf;
+		quoting = 0;
+		while(*q!='\0' && *q!='\n'){
+			if(v >= attrbuf+sizeof attrbuf)
+				break;
+			c = *q++;
+			if(quoting){
+				if(c == '\''){
+					if(*q == '\'')
+						q++;
+					else{
+						quoting = 0;
+						continue;
+					}
+				}
+			}else{
+				if(c==' ' || c=='\t')
+					break;
+				if(c == '\''){
+					quoting = 1;
+					continue;
+				}
+			}
+			*v++ = c;
+		}
+		a->value = malloc(v-attrbuf+1);
+		if(a->value == nil){
+			free(a->name);
+			free(a);
+			break;
+		}
+		memmove(a->value, attrbuf, v-attrbuf);
+		a->value[v-attrbuf] = '\0';
+		a->next = nil;
+		if(prev == nil)
+			attr = a;
+		else
+			prev->next = a;
+		prev = a;
+		p = q;
+	}
+	return attr;
+}
+
+Plumbattr*
+plumbaddattr(Plumbattr *attr, Plumbattr *new)
+{
+	Plumbattr *l;
+
+	l = attr;
+	if(l == nil)
+		return new;
+	while(l->next != nil)
+		l = l->next;
+	l->next = new;
+	return attr;
+}
+
+Plumbattr*
+plumbdelattr(Plumbattr *attr, char *name)
+{
+	Plumbattr *l, *prev;
+
+	prev = nil;
+	for(l=attr; l!=nil; l=l->next){
+		if(strcmp(name, l->name) == 0)
+			break;
+		prev = l;
+	}
+	if(l == nil)
+		return nil;
+	if(prev)
+		prev->next = l->next;
+	else
+		attr = l->next;
+	free(l->name);
+	free(l->value);
+	free(l);
+	return attr;
+}
+
+Plumbmsg*
+plumbunpackpartial(char *buf, int n, int *morep)
+{
+	Plumbmsg *m;
+	int i, bad;
+	char *ntext, *attr;
+
+	m = malloc(sizeof(Plumbmsg));
+	if(m == nil)
+		return nil;
+	memset(m, 0, sizeof(Plumbmsg));
+	if(morep != nil)
+		*morep = 0;
+	bad = 0;
+	i = plumbline(&m->src, buf, 0, n, &bad);
+	i = plumbline(&m->dst, buf, i, n, &bad);
+	i = plumbline(&m->wdir, buf, i, n, &bad);
+	i = plumbline(&m->type, buf, i, n, &bad);
+	i = plumbline(&attr, buf, i, n, &bad);
+	m->attr = plumbunpackattr(attr);
+	free(attr);
+	i = plumbline(&ntext, buf, i, n, &bad);
+	m->ndata = atoi(ntext);
+	if(m->ndata != n-i){
+		bad = 1;
+		if(morep!=nil && m->ndata>n-i)
+			*morep = m->ndata - (n-i);
+	}
+	free(ntext);
+	if(!bad){
+		m->data = malloc(n-i+1);	/* +1 for '\0' */
+		if(m->data == nil)
+			bad = 1;
+		else{
+			memmove(m->data, buf+i, m->ndata);
+			m->ndata = n-i;
+			/* null-terminate in case it's text */
+			m->data[m->ndata] = '\0';
+		}
+	}
+	if(bad){
+		plumbfree(m);
+		m = nil;
+	}
+	return m;
+}
+
+Plumbmsg*
+plumbunpack(char *buf, int n)
+{
+	return plumbunpackpartial(buf, n, nil);
+}
+
+Plumbmsg*
+plumbrecv(int fd)
+{
+	char *buf;
+	Plumbmsg *m;
+	int n, more;
+
+	buf = malloc(8192);
+	if(buf == nil)
+		return nil;
+	n = read(fd, buf, 8192);
+	m = nil;
+	if(n > 0){
+		m = plumbunpackpartial(buf, n, &more);
+		if(m==nil && more>0){
+			/* we now know how many more bytes to read for complete message */
+			buf = realloc(buf, n+more);
+			if(buf == nil)
+				return nil;
+			if(readn(fd, buf+n, more) == more)
+				m = plumbunpackpartial(buf, n+more, nil);
+		}
+	}
+	free(buf);
+	return m;
+}