WIP: transition to eio

src/lib/client/dune

@@ -28,7 +28,8 @@
   logs.browser
   cohttp
   tyxml
-  reactiveData)
+  reactiveData
+  eio)
  (foreign_stubs
   (language c)
   (names eliom_stubs))

src/lib/client/eliommod_dom.ml

@@ -1,4 +1,4 @@
-open Lwt.Syntax
+open Eio.Std
 
 (* Ocsigen
  * http://www.ocsigen.org
@@ -479,7 +479,7 @@ let fetch_linked_css e =
           let css =
             Eliom_request.http_get href [] Eliom_request.string_result
           in
-          acc @ [e, (e##.media, href, css >|= snd)]
+          acc @ [e, (e##.media, href, snd css)]
     | Dom.Element e ->
         let c = e##.childNodes in
         let acc = ref acc in
@@ -578,26 +578,25 @@ let rewrite_css_url ~prefix css pos =
 let import_re = Regexp.regexp "@import\\s*"
 
 let rec rewrite_css ~max (media, href, css) =
-  Lwt.catch
-    (fun () ->
-       css >>= function
-       | None -> Lwt.return_nil
-       | Some css ->
-           if !Eliom_config.debug_timings
-           then Console.console##(time (Js.string ("rewrite_CSS: " ^ href)));
-           let* imports, css =
-             rewrite_css_import ~max ~prefix:(basedir href) ~media css 0
-           in
-           if !Eliom_config.debug_timings
-           then Console.console##(timeEnd (Js.string ("rewrite_CSS: " ^ href)));
-           Lwt.return (imports @ [media, css]))
-    (fun _ -> Lwt.return [media, Printf.sprintf "@import url(%s);" href])
+  try
+    match css with
+    | None -> []
+    | Some css ->
+        if !Eliom_config.debug_timings
+        then Console.console##(time (Js.string ("rewrite_CSS: " ^ href)));
+        let imports, css =
+          rewrite_css_import ~max ~prefix:(basedir href) ~media css 0
+        in
+        if !Eliom_config.debug_timings
+        then Console.console##(timeEnd (Js.string ("rewrite_CSS: " ^ href)));
+        imports @ [media, css]
+  with _ -> [media, Printf.sprintf "@import url(%s);" href]
 
 and rewrite_css_import ?(charset = "") ~max ~prefix ~media css pos =
   match Regexp.search import_re css pos with
   | None ->
       (* No @import anymore, rewrite url. *)
-      Lwt.return ([], rewrite_css_url ~prefix css pos)
+      [], rewrite_css_url ~prefix css pos
   | Some (i, res) -> (
       (* Found @import rule, try to preload. *)
       let init = String.sub css pos (i - pos) in
@@ -606,45 +605,46 @@ and rewrite_css_import ?(charset = "") ~max ~prefix ~media css pos =
         let i = i + String.length (Regexp.matched_string res) in
         let i, href = parse_url ~prefix css i in
         let i, media' = parse_media css i in
-        let* import =
-          if max = 0
-          then
-            (* Maximum imbrication of @import reached, rewrite url. *)
-            Lwt.return
-              [media, Printf.sprintf "@import url('%s') %s;\n" href media']
-          else if media##.length > 0 && String.length media' > 0
-          then
-            (* TODO combine media if possible...
+        let (imports, css), import =
+          Fiber.pair
+            (fun () -> rewrite_css_import ~charset ~max ~prefix ~media css i)
+            (fun () ->
+               if
+                 (* TODO: lwt-to-direct-style: This computation might not be suspended correctly. *)
+                 max = 0
+               then
+                 (* Maximum imbrication of @import reached, rewrite url. *)
+                 [media, Printf.sprintf "@import url('%s') %s;\n" href media']
+               else if media##.length > 0 && String.length media' > 0
+               then
+                 (* TODO combine media if possible...
                in the mean time keep explicit import. *)
-            Lwt.return
-              [media, Printf.sprintf "@import url('%s') %s;\n" href media']
-          else
-            let media =
-              if media##.length > 0 then media else Js.string media'
-            in
-            let css =
-              Eliom_request.http_get href [] Eliom_request.string_result
-            in
-            rewrite_css ~max:(max - 1) (media, href, css >|= snd)
-        and* imports, css =
-          rewrite_css_import ~charset ~max ~prefix ~media css i
+                 [media, Printf.sprintf "@import url('%s') %s;\n" href media']
+               else
+                 let media =
+                   if media##.length > 0 then media else Js.string media'
+                 in
+                 let css =
+                   Eliom_request.http_get href [] Eliom_request.string_result
+                 in
+                 rewrite_css ~max:(max - 1) (media, href, snd css))
         in
-        Lwt.return (import @ imports, css)
+        import @ imports, css
       with
-      | Incorrect_url -> Lwt.return ([], rewrite_css_url ~prefix css pos)
+      | Incorrect_url -> [], rewrite_css_url ~prefix css pos
       | exn ->
           Logs.info ~src:section (fun fmt ->
             fmt
               ("Error while importing css" ^^ "@\n%s")
               (Printexc.to_string exn));
-          Lwt.return ([], rewrite_css_url ~prefix css pos))
+          [], rewrite_css_url ~prefix css pos)
 
 let max_preload_depth = ref 4
 
 let build_style (e, css) =
-  let* css = rewrite_css ~max:!max_preload_depth css in
-  (* lwt css = *)
-  Lwt_list.map_p
+  let css = rewrite_css ~max:!max_preload_depth css in
+  Fiber.List.map
+    (* lwt css = *)
     (fun (media, css) ->
        let style = Dom_html.createStyle Dom_html.document in
        style##._type := Js.string "text/css";
@@ -655,7 +655,7 @@ let build_style (e, css) =
        if Js.Optdef.test styleSheet
        then Js.Unsafe.(set styleSheet (Js.string "cssText") (Js.string css))
        else style##.innerHTML := Js.string css;
-       Lwt.return (e, (style :> Dom.node Js.t)))
+       e, (style :> Dom.node Js.t))
     css
 
 (* IE8 doesn't allow appendChild on noscript-elements *)
@@ -669,7 +669,7 @@ let build_style (e, css) =
 let preload_css (doc : Dom_html.element Js.t) =
   if !Eliom_config.debug_timings
   then Console.console##(time (Js.string "preload_css (fetch+rewrite)"));
-  let* css = Lwt_list.map_p build_style (fetch_linked_css (get_head doc)) in
+  let css = Fiber.List.map build_style (fetch_linked_css (get_head doc)) in
   let css = List.concat css in
   List.iter
     (fun (e, css) ->
@@ -682,8 +682,7 @@ let preload_css (doc : Dom_html.element Js.t) =
              section (fun fmt -> fmt "Unique CSS skipped..."))
     css;
   if !Eliom_config.debug_timings
-  then Console.console##(timeEnd (Js.string "preload_css (fetch+rewrite)"));
-  Lwt.return_unit
+  then Console.console##(timeEnd (Js.string "preload_css (fetch+rewrite)"))
 
 (** Window scrolling *)
 

src/lib/client/eliommod_dom.mli

@@ -71,7 +71,7 @@ val html_document :
 (** Assuming [d] has a body and head element, [html_document d] will
     return the same document as html *)
 
-val preload_css : Dom_html.element Js.t -> unit Lwt.t
+val preload_css : Dom_html.element Js.t -> unit
 (** [preload_css e] downloads every css included in every link
     elements that is a descendant of [e] and replace it and its linked
     css by inline [<style>] elements *)

src/lib/eliom_bus.client.ml

@@ -1,4 +1,4 @@
-open Lwt.Syntax
+open Eio.Std
 
 (* Ocsigen
  * http://www.ocsigen.org
@@ -27,36 +27,16 @@ let section = Logs.Src.create "eliom:bus"
 
 module Ecb = Eliom_comet_base
 
+type 'a consumers = {mutable consumers : ('a option -> unit) list}
+
 type ('a, 'b) t =
   { channel : 'b Ecb.wrapped_channel
-  ; stream : 'b Lwt_stream.t Lazy.t
+  ; consumers : 'b consumers Lazy.t
   ; queue : 'a Queue.t
   ; mutable max_size : int
-  ; write : 'a list -> unit Lwt.t
-  ; mutable waiter : unit -> unit Lwt.t
-  ; mutable last_wait : unit Lwt.t
-  ; mutable original_stream_available : bool
-  ; error_h : 'b option Lwt.t * exn Lwt.u }
-
-(* clone streams such that each clone of the original stream raise the same exceptions *)
-let consume (t, u) s =
-  let t' =
-    Lwt.catch
-      (fun () -> Lwt_stream.iter (fun _ -> ()) s)
-      (fun e ->
-         (match Lwt.state t with Lwt.Sleep -> Lwt.wakeup_exn u e | _ -> ());
-         Lwt.fail e)
-  in
-  Lwt.choose [Lwt.bind t (fun _ -> Lwt.return_unit); t']
-
-let clone_exn (t, u) s =
-  let s' = Lwt_stream.clone s in
-  Lwt_stream.from (fun () ->
-    Lwt.catch
-      (fun () -> Lwt.choose [Lwt_stream.get s'; t])
-      (fun e ->
-         (match Lwt.state t with Lwt.Sleep -> Lwt.wakeup_exn u e | _ -> ());
-         Lwt.fail e))
+  ; write : 'a list -> unit
+  ; mutable waiter : unit -> unit
+  ; mutable last_wait : unit Promise.t }
 
 type ('a, 'att, 'co, 'ext, 'reg) callable_bus_service =
   ( unit
@@ -72,57 +52,47 @@ type ('a, 'att, 'co, 'ext, 'reg) callable_bus_service =
     , Eliom_registration.Action.return )
     Eliom_service.t
 
-let create service channel waiter =
-  let write x =
-    Lwt.catch
-      (fun () ->
-         let* _ =
-           Eliom_client.call_service
-             ~service:(service :> ('a, _, _, _, _) callable_bus_service)
-             () x
-         in
-         Lwt.return_unit)
-      (function
-        | Eliom_request.Failed_request 204 -> Lwt.return_unit
-        | exc -> Lwt.reraise exc)
-  in
-  let error_h =
-    let t, u = Lwt.wait () in
-    ( Lwt.catch
-        (fun () ->
-           let* _ = t in
-           assert false)
-        (fun e -> Lwt.fail e)
-    , u )
-  in
-  let stream =
-    lazy
-      (let stream = Eliom_comet.register channel in
-       (* iterate on the stream to consume messages: avoid memory leak *)
-       let _ = consume error_h stream in
-       stream)
+(** Register a callback in the underlying comet. *)
+let comet_register chan =
+  let t = {consumers = []} in
+  let notify data = List.iter (fun callback -> callback data) t.consumers in
+  let teardown () =
+    let
+        (* Notify that the channel reached its end. Clear the [consumers] list to
+       avoid memory leaks. *)
+          ()
+      =
+      notify None
+    in
+    t.consumers <- []
   in
-  let t =
-    { channel
-    ; stream
-    ; queue = Queue.create ()
-    ; max_size = 20
-    ; write
-    ; waiter
-    ; last_wait = Lwt.return_unit
-    ; original_stream_available = true
-    ; error_h }
-  in
-  (* the comet channel start receiving after the load phase, so the
-     original channel (i.e. without message lost) is only available in
-     the first loading phase. *)
-  let _ =
-    let* () = Eliom_client.wait_load_end () in
-    t.original_stream_available <- false;
-    Lwt.return_unit
+  let _chan =
+    Eliom_comet.register_wrapped chan (function
+      | Some data -> notify (Some data)
+      | None -> teardown ())
   in
   t
 
+let create service channel waiter =
+  let write x =
+    try
+      let _ =
+        Eliom_client.call_service
+          ~service:(service :> ('a, _, _, _, _) callable_bus_service)
+          () x
+      in
+      ()
+    with Eliom_request.Failed_request 204 -> ()
+  in
+  let consumers = lazy (comet_register channel) in
+  { channel
+  ; consumers
+  ; queue = Queue.create ()
+  ; max_size = 20
+  ; write
+  ; waiter
+  ; last_wait = () }
+
 let internal_unwrap ((wrapped_bus : ('a, 'b) Ecb.wrapped_bus), _unwrapper) =
   let waiter () = Js_of_ocaml_lwt.Lwt_js.sleep 0.05 in
   let channel, Eliom_comet_base.Bus_send_service service = wrapped_bus in
@@ -131,35 +101,40 @@ let internal_unwrap ((wrapped_bus : ('a, 'b) Ecb.wrapped_bus), _unwrapper) =
 let () =
   Eliom_unwrap.register_unwrapper Eliom_common.bus_unwrap_id internal_unwrap
 
-let stream t = clone_exn t.error_h (Lazy.force t.stream)
-
-let original_stream t =
-  if Eliom_client_core.in_onload () && t.original_stream_available
-  then stream t
-  else
-    raise_error ~section
-      "original_stream: the original stream is not available anymore"
+let register t callback =
+  let (lazy c) = t.consumers in
+  c.consumers <- callback :: c.consumers
 
 let flush t =
   let l = List.rev (Queue.fold (fun l v -> v :: l) [] t.queue) in
   Queue.clear t.queue; t.write l
 
 let try_flush t =
-  Lwt.cancel t.last_wait;
+  Lwt.cancel
+    (* TODO: lwt-to-direct-style: Use [Switch] or [Cancel] for defining a cancellable context. *)
+    (* TODO: lwt-to-direct-style: Use [Switch] or [Cancel] for defining a cancellable context. *)
+    t.last_wait;
   if Queue.length t.queue >= t.max_size
   then flush t
   else
-    let th = Lwt.protected (t.waiter ()) in
+    let th =
+      Lwt.protected
+        (* TODO: lwt-to-direct-style: Use [Switch] or [Cancel] for defining a cancellable context. *)
+        (* TODO: lwt-to-direct-style: Use [Switch] or [Cancel] for defining a cancellable context. *)
+        (t.waiter ())
+    in
     t.last_wait <- th;
-    let _ = th >>= fun () -> flush t in
-    Lwt.return_unit
+    let _ = th; flush t in
+    ()
 
 let write t v = Queue.add v t.queue; try_flush t
 let close {channel; _} = Eliom_comet.close channel
 let set_queue_size b s = b.max_size <- s
 
 let set_time_before_flush b t =
   b.waiter <-
-    (if t <= 0. then Lwt.pause else fun () -> Js_of_ocaml_lwt.Lwt_js.sleep t)
+    (if t <= 0.
+     then fun x1 -> Fiber.yield x1
+     else fun () -> Js_of_ocaml_lwt.Lwt_js.sleep t)
 
 let force_link = ()