Redo channels, making them single-client

4 files changed, 69 insertions, 0 deletions

diff --git a/lib/js/urweb.js b/lib/js/urweb.js
index c7cecbbf..4daee66d 100644
--- a/lib/js/urweb.js
+++ b/lib/js/urweb.js
@@ -301,6 +301,9 @@ function listener() {
 }
 
 function rv(chn, parse, k) {
+  if (chn == null)
+    return;
+
   if (chn < 0)
     whine("Out-of-bounds channel receive");
 
diff --git a/lib/ur/basis.urs b/lib/ur/basis.urs
index 23c3fe57..d6b27852 100644
--- a/lib/ur/basis.urs
+++ b/lib/ur/basis.urs
@@ -115,6 +115,9 @@ 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
 
+type client
+val self : transaction client
+
 
 (** SQL *)
 
@@ -207,6 +210,7 @@ val sql_time : sql_injectable_prim time
 
 class sql_injectable_nullable
 val sql_channel : t ::: Type -> sql_injectable_nullable (channel t)
+val sql_client : sql_injectable_nullable client
 
 class sql_injectable
 val sql_prim : t ::: Type -> sql_injectable_prim t -> sql_injectable t
diff --git a/lib/ur/top.ur b/lib/ur/top.ur
index 713f4311..5465da3d 100644
--- a/lib/ur/top.ur
+++ b/lib/ur/top.ur
@@ -143,6 +143,14 @@ fun foldRX2 K (tf1 :: K -> Type) (tf2 :: K -> Type) (ctx :: {Unit})
                <xml>{f [nm] [t] [rest] ! r1 r2}{acc}</xml>)
            <xml/>
 
+fun queryI (tables ::: {{Type}}) (exps ::: {Type})
+           [tables ~ exps] (q : sql_query tables exps)
+           (f : $(exps ++ map (fn fields :: {Type} => $fields) tables)
+                -> transaction unit) =
+    query q
+          (fn fs _ => f fs)
+          ()
+
 fun queryX (tables ::: {{Type}}) (exps ::: {Type}) (ctx ::: {Unit})
            [tables ~ exps] (q : sql_query tables exps)
            (f : $(exps ++ map (fn fields :: {Type} => $fields) tables)
@@ -188,3 +196,39 @@ fun eqNullable' (tables ::: {{Type}}) (agg ::: {{Type}}) (exps ::: {Type})
     case e2 of
         None => (SQL {e1} IS NULL)
       | Some _ => sql_binary sql_eq e1 (sql_inject e2)
+
+
+functor Broadcast(M : sig type t end) = struct
+    sequence s
+    table t : {Id : int, Client : option client, Channel : option (channel M.t)}
+
+    type topic = int
+
+    val inj : sql_injectable topic = _
+
+    val create = nextval s
+
+    val cleanup =
+        dml (DELETE FROM t WHERE Client IS NULL)
+
+    fun subscribe id =
+        cli <- self;
+        cleanup;
+        ro <- oneOrNoRows (SELECT t.Channel FROM t WHERE t.Id = {[id]} AND t.Client = {[Some cli]});
+        case ro of
+            None =>
+            ch <- channel;
+            dml (INSERT INTO t (Id, Client, Channel) VALUES ({[id]}, {[Some cli]}, {[Some ch]}));
+            return ch
+          | Some r =>
+            case r.T.Channel of
+                None => error <xml>Broadcast.subscribe: Got null result</xml>
+              | Some ch => return ch
+
+    fun send id msg =
+        cleanup;
+        queryI (SELECT t.Channel FROM t WHERE t.Id = {[id]})
+        (fn r => case r.T.Channel of
+                     None => error <xml>Broadcast.send: Got null result</xml>
+                   | Some ch => Basis.send ch msg)
+end
diff --git a/lib/ur/top.urs b/lib/ur/top.urs
index bc0b768e..821aa42a 100644
--- a/lib/ur/top.urs
+++ b/lib/ur/top.urs
@@ -87,6 +87,13 @@ val foldRX2 : K --> tf1 :: (K -> Type) -> tf2 :: (K -> Type) -> ctx :: {Unit}
               -> r :: {K} -> folder r
               -> $(map tf1 r) -> $(map tf2 r) -> xml ctx [] []
 
+val queryI : tables ::: {{Type}} -> exps ::: {Type}
+             -> [tables ~ exps] =>
+             sql_query tables exps
+             -> ($(exps ++ map (fn fields :: {Type} => $fields) tables)
+                 -> transaction unit)
+             -> transaction unit
+
 val queryX : tables ::: {{Type}} -> exps ::: {Type} -> ctx ::: {Unit}
              -> [tables ~ exps] =>
              sql_query tables exps
@@ -127,3 +134,14 @@ val eqNullable' : tables ::: {{Type}} -> agg ::: {{Type}} -> exps ::: {Type}
                   -> sql_exp tables agg exps (option t)
                   -> option t
                   -> sql_exp tables agg exps bool
+
+
+functor Broadcast(M : sig type t end) : sig
+    type topic
+
+    val inj : sql_injectable topic
+
+    val create : transaction topic
+    val subscribe : topic -> transaction (channel M.t)
+    val send : topic -> M.t -> transaction unit
+end