Proper recv

2 files changed, 77 insertions, 2 deletions

diff --git a/lib/js/urweb.js b/lib/js/urweb.js
index 18842188..6cb5c60a 100644
--- a/lib/js/urweb.js
+++ b/lib/js/urweb.js
@@ -1,6 +1,7 @@
 function cons(v, ls) {
   return { n : ls, v : v };
 }
+
 function callAll(ls) {
   for (; ls; ls = ls.n)
     ls.v();
@@ -192,7 +193,6 @@ function rc(uri, parse, k) {
   requestUri(xhr, uri);
 }
 
-
 function path_join(s1, s2) {
   if (s1.length > 0 && s1[s1.length-1] == '/')
     return s1 + s2;
@@ -200,6 +200,37 @@ function path_join(s1, s2) {
     return s1 + "/" + s2;
 }
 
+var channels = [];
+
+function newQueue() {
+  return { front : null, back : null };
+}
+function enqueue(q, v) {
+  if (q.front == null) {
+    q.front = cons(v, null);
+    q.back = q.front;
+  } else {
+    var node = cons(v, null);
+    q.back.n = node;
+    q.back = node;
+  }
+}
+function dequeue(q) {
+  if (q.front == null)
+    return null;
+  else {
+    var r = q.front.v;
+    q.front = q.front.n;
+    if (q.front == null)
+      q.back = null;
+    return r;
+  }
+}
+
+function newChannel() {
+  return { msgs : newQueue(), listeners : newQueue() };
+}
+
 function listener() {
   var uri = path_join(url_prefix, ".msgs");
   var xhr = getXHR();
@@ -218,7 +249,26 @@ function listener() {
           whine("Empty message from remote server");
 
         for (var i = 0; i+1 < lines.length; i += 2) {
-          alert("Message(" + lines[i] + "): " + lines[i+1]);
+          var chn = lines[i];
+          var msg = lines[i+1];
+
+          if (chn < 0)
+            whine("Out-of-bounds channel in message from remote server");
+
+          var ch;
+
+          if (chn >= channels.length || channels[chn] == null) {
+            ch = newChannel();
+            channels[chn] = ch;
+          } else
+            ch = channels[chn];
+
+          var listener = dequeue(ch.listeners);
+          if (listener == null) {
+            enqueue(ch.msgs, msg);
+          } else {
+            listener(msg);
+          }
         }
 
         xhr.onreadystatechange = orsc;
@@ -233,3 +283,27 @@ function listener() {
   xhr.onreadystatechange = orsc;
   requestUri(xhr, uri);
 }
+
+function rv(chn, parse, k) {
+  if (chn < 0)
+    whine("Out-of-bounds channel receive");
+
+  var ch;
+
+  if (chn >= channels.length || channels[chn] == null) {
+    ch = newChannel();
+    channels[chn] = ch;
+  } else
+    ch = channels[chn];
+
+  var msg = dequeue(ch.msgs);
+  if (msg == null) {
+    enqueue(ch.listeners, function(msg) { k(parse(msg))(null); });
+  } else {
+    k(parse(msg))(null);
+  }
+}
+
+function unesc(s) {
+  return unescape(s).replace("+", " ");
+}
diff --git a/lib/ur/basis.urs b/lib/ur/basis.urs
index e7172db1..8c28dacb 100644
--- a/lib/ur/basis.urs
+++ b/lib/ur/basis.urs
@@ -460,3 +460,4 @@ con channel :: Type -> Type
 val channel : t ::: Type -> transaction (channel t)
 val subscribe : t ::: Type -> channel t -> transaction unit
 val send : t ::: Type -> channel t -> t -> transaction unit
+val recv : t ::: Type -> channel t -> transaction t