explicitly cancellable notifications

src/froc/froc.ml

@@ -64,13 +64,22 @@ let send_result s r =
 let send s v = send_result s (Value v)
 let send_exn s e = send_result s (Fail e)
 
-let notify_result_e t f =
+let notify_result_e_cancel t f =
   match repr_of_event t with
-    | Never -> ()
+    | Never -> ignore
     | Occurs o ->
         let dl = Dlist.add_after o.e_deps f in
-        let cancel () = Dlist.remove dl in
-        TS.add_cleanup (TS.tick ()) cancel
+        fun () -> Dlist.remove dl
+
+let notify_result_e t f =
+  let cancel = notify_result_e_cancel t f in
+  TS.add_cleanup (TS.tick ()) cancel
+
+let notify_e_cancel t f =
+  notify_result_e_cancel t
+    (function
+       | Value v -> f v
+       | Fail _ -> ())
 
 let notify_e t f =
   notify_result_e t
@@ -158,8 +167,9 @@ let send_exn t e = send_result t (Fail e)
 type 'a behavior = 'a t
 
 let notify_b = notify
-
+let notify_b_cancel = notify_cancel
 let notify_result_b = notify_result
+let notify_result_b_cancel = notify_result_cancel
 
 let hash_behavior = hash
 

src/froc/froc.mli

@@ -136,11 +136,25 @@ val read_result : 'a behavior -> 'a result
      an exception.
   *)
 
+type cancel
+  (** Type of handles to listener registrations. *)
+
+val cancel : cancel -> unit
+  (** Cancels a listener registration using the given handle. *)
+
+
 val notify_b : 'a behavior -> ('a -> unit) -> unit
   (**
      Adds a listener for the value of a behavior, which is called
      whenever the value changes. When the behavior fails the listener
-     is not called.
+     is not called. The notification is implicitly cancelled when the
+     calling context is re-run.
+  *)
+
+val notify_b_cancel : 'a behavior -> ('a -> unit) -> cancel
+  (**
+     Same as [notify_b], but not implicitly cancelled; an explicit
+     cancel handle is returned.
   *)
 
 val notify_result_b : 'a behavior -> ('a result -> unit) -> unit
@@ -149,6 +163,12 @@ val notify_result_b : 'a behavior -> ('a result -> unit) -> unit
      the value changes or when the behavior fails.
   *)
 
+val notify_result_b_cancel : 'a behavior -> ('a result -> unit) -> cancel
+  (**
+     Same as [notify_b_cancel] but the listener is called with a
+     result when the value changes or when the behavior fails.
+  *)
+
 val cleanup : (unit -> unit) -> unit
   (**
      When called in the context of a binder, adds a function to be
@@ -201,7 +221,15 @@ val make_event : unit -> 'a event * 'a event_sender
 val notify_e : 'a event -> ('a -> unit) -> unit
   (**
      Adds a listener on the channel, which is called whenever a value
-     is sent on it. When a failure is sent the listener is not called.
+     is sent on it. When a failure is sent the listener is not
+     called. The notification is implicitly cancelled when the calling
+     context is re-run.
+  *)
+
+val notify_e_cancel : 'a event -> ('a -> unit) -> cancel
+  (**
+     Same as [notify_e], but not implicitly cancelled; an explicit
+     cancel handle is returned.
   *)
 
 val notify_result_e : 'a event -> ('a result -> unit) -> unit
@@ -210,6 +238,12 @@ val notify_result_e : 'a event -> ('a result -> unit) -> unit
      a value or a failure is sent.
   *)
 
+val notify_result_e_cancel : 'a event -> ('a result -> unit) -> cancel
+  (**
+     Same as [notify_e_cancel] but the listener is called with a result when
+     a value or a failure is sent.
+  *)
+
 val send : 'a event_sender -> 'a -> unit
   (** [send e v] calls the listeners of the associated event with [Value v]. *) 
 

src/froc/froc_ddg.ml

@@ -116,17 +116,32 @@ let read t =
     | Value v -> v
     | Fail e -> raise e
 
-let add_dep ts t dep =
+type cancel = unit -> unit
+let cancel c = c ()
+
+let add_dep_cancel t f =
   match repr_of_t t with
-    | Constant _ -> ()
+    | Constant _ -> ignore
     | Changeable c ->
-        let dl = Dlist.add_after c.deps dep in
-        let cancel () = Dlist.remove dl in
-        TS.add_cleanup ts cancel
+        let dl = Dlist.add_after c.deps f in
+        fun () -> Dlist.remove dl
+
+let add_dep ts t dep =
+  let cancel = add_dep_cancel t dep in
+  TS.add_cleanup ts cancel
+
+let notify_result_cancel t f =
+  add_dep_cancel t f
 
 let notify_result t f =
   add_dep (TS.tick ()) t f
 
+let notify_cancel t f =
+  notify_result_cancel t
+    (function
+       | Fail _ -> ()
+       | Value v -> f v)
+
 let notify t f =
   notify_result t
     (function
@@ -250,6 +265,10 @@ let add_reader t read =
   let dep _ = enqueue r in
   add_dep start t dep
 
+let connect_cancel u t' =
+  write_result u (read_result t');
+  add_dep_cancel t' (write_result_no_eq u)
+
 let connect u t' =
   write_result u (read_result t');
   add_dep (TS.tick ()) t' (write_result_no_eq u)

src/froc/froc_ddg.mli

@@ -47,8 +47,17 @@ val write_result : 'a u -> 'a result -> unit
 
 val notify : 'a t -> ('a -> unit) -> unit
 val notify_result : 'a t -> ('a result -> unit) -> unit
+val connect : 'a u -> 'a t -> unit
 val cleanup : (unit -> unit) -> unit
 
+type cancel = unit -> unit
+
+val notify_cancel : 'a t -> ('a -> unit) -> cancel
+val notify_result_cancel : 'a t -> ('a result -> unit) -> cancel
+val connect_cancel : 'a u -> 'a t -> cancel
+
+val cancel : cancel -> unit
+
 val make_changeable : ?eq:('a -> 'a -> bool) -> ?result:'a result -> unit -> 'a t * 'a u
 val make_constant : 'a result -> 'a t