feat: implement download and websocket with httpun

dune-project

@@ -33,7 +33,6 @@
   fmt
   libspng
   odiff-core
-  piaf
   re
   uri
   yaml

lib/OSnap_Browser/OSnap_Browser_Download.ml

@@ -1,70 +1,76 @@
 open OSnap_Utils
 
 let get_uri revision (platform : OSnap_Utils.platform) =
-  Uri.make
-    ~scheme:"http"
-    ~host:"storage.googleapis.com"
-    ~port:80
-    ~path:
-      (match platform with
-       | MacOS -> "/chromium-browser-snapshots/Mac/" ^ revision ^ "/chrome-mac.zip"
-       | MacOS_ARM ->
-         "/chromium-browser-snapshots/Mac_Arm/" ^ revision ^ "/chrome-mac.zip"
-       | Linux ->
-         "/chromium-browser-snapshots/Linux_x64/" ^ revision ^ "/chrome-linux.zip"
-       | Win64 -> "/chromium-browser-snapshots/Win_x64/" ^ revision ^ "/chrome-win.zip"
-       | Win32 ->
-         print_endline "Error: x86 is currently not supported on Windows";
-         exit 1)
-    ()
+  let host = "storage.googleapis.com" in
+  let port = 80 in
+  let path =
+    match platform with
+    | MacOS -> "/chromium-browser-snapshots/Mac/" ^ revision ^ "/chrome-mac.zip"
+    | MacOS_ARM -> "/chromium-browser-snapshots/Mac_Arm/" ^ revision ^ "/chrome-mac.zip"
+    | Linux -> "/chromium-browser-snapshots/Linux_x64/" ^ revision ^ "/chrome-linux.zip"
+    | Win64 -> "/chromium-browser-snapshots/Win_x64/" ^ revision ^ "/chrome-win.zip"
+    | Win32 ->
+      print_endline "Error: x86 is currently not supported on Windows";
+      exit 1
+  in
+  host, port, path
 ;;
 
-let download ~env ~revision dir =
-  let ( let*? ) = Result.bind in
+let download ~revision dir =
   let zip_path = Eio.Path.(dir / "chromium.zip") in
+  Eio.Path.rmtree ~missing_ok:true zip_path;
   let revision_string = OSnap_Browser_Path.revision_to_string revision in
-  let uri = get_uri revision_string (OSnap_Utils.detect_platform ()) in
+  let host, port, path = get_uri revision_string (OSnap_Utils.detect_platform ()) in
   print_endline
     (Printf.sprintf
        "Downloading chromium revision %s.\n\
         This is a one time setup and will only happen, if there are updates from OSnap..."
        revision_string);
   Eio.Switch.run
   @@ fun sw ->
-  let*? client =
-    Piaf.Client.create
-      ~sw
-      ~config:
-        { Piaf.Config.default with
-          follow_redirects = true
-        ; allow_insecure = true
-        ; flush_headers_immediately = true
-        }
-      env
-      uri
-    |> Result.map_error (fun _ -> `OSnap_Chromium_Download_Failed)
-  in
-  Eio.Path.with_open_out ~create:(`Or_truncate 0o755) zip_path
-  @@ fun io ->
-  let*? response =
-    Piaf.Client.get client (Uri.path_and_query uri)
-    |> Result.map_error (fun _ -> `OSnap_Chromium_Download_Failed)
+  let fd = Unix.socket ~cloexec:true Unix.PF_INET Unix.SOCK_STREAM 0 in
+  let addrs =
+    Eio_unix.run_in_systhread (fun () ->
+      Unix.getaddrinfo host (Int.to_string port) [ Unix.(AI_FAMILY PF_INET) ])
   in
-  match response with
-  | { status = `OK; _ } ->
-    let*? () =
-      Piaf.Body.iter_string ~f:(fun chunk -> Eio.Flow.copy_string chunk io) response.body
-      |> Result.map_error (fun _ -> `OSnap_Chromium_Download_Failed)
+  Eio_unix.run_in_systhread (fun () -> Unix.connect fd (List.hd addrs).ai_addr);
+  let socket = Eio_unix.Net.import_socket_stream ~sw ~close_unix:true fd in
+  let headers = Httpun.Headers.of_list [ "host", host ] in
+  let connection = Httpun_eio.Client.create_connection ~sw socket in
+  let p, resolver = Eio.Promise.create () in
+  Eio.Fiber.fork ~sw (fun () ->
+    let request_body =
+      Httpun_eio.Client.request
+        connection
+        (Httpun.Request.create ~headers `GET path)
+        ~error_handler:(fun _ -> ())
+        ~response_handler:(fun response body ->
+          let buf = Buffer.create (1024 * 1024 * 1024 * 140) in
+          match response with
+          | { Httpun.Response.status = `OK; _ } ->
+            let on_eof () = Eio.Promise.resolve_ok resolver (Buffer.contents buf) in
+            let rec on_read bs ~off ~len =
+              let chunk = Bigstringaf.substring ~off ~len bs in
+              Eio.Path.with_open_out
+                ~append:true
+                ~create:(`If_missing 0o755)
+                zip_path
+                (Eio.Flow.copy_string chunk);
+              Httpun.Body.Reader.schedule_read body ~on_read ~on_eof
+            in
+            Httpun.Body.Reader.schedule_read body ~on_read ~on_eof
+          | response ->
+            Format.fprintf
+              Format.err_formatter
+              "Chrome could not be downloaded:\n%a\n%!"
+              Httpun.Response.pp_hum
+              response;
+            Eio.Promise.resolve_error resolver `OSnap_Chromium_Download_Failed)
     in
-    Piaf.Client.shutdown client;
-    Result.ok zip_path
-  | response ->
-    Format.fprintf
-      Format.err_formatter
-      "Chrome could not be downloaded:\n%a\n%!"
-      Piaf.Response.pp_hum
-      response;
-    Result.error `OSnap_Chromium_Download_Failed
+    Httpun.Body.Writer.close request_body);
+  let*? _response_body = Eio.Promise.await p in
+  Httpun_eio.Client.shutdown connection |> Eio.Promise.await;
+  Result.ok zip_path
 ;;
 
 let extract_zip ~dest source =
@@ -136,7 +142,7 @@ let download ~env revision =
       Eio.Path.with_open_dir
         Eio.Path.(fs / temp_dir)
         (fun dir ->
-          let*? path = download dir ~env ~revision in
+          let*? path = download dir ~revision in
           extract_zip path ~dest:extract_path;
           Eio.Path.rmtree ~missing_ok:true path;
           Result.ok ()))

lib/OSnap_Browser/OSnap_Browser_Launcher.ml

@@ -76,7 +76,7 @@ let make ~sw ~env () =
   in
   let stderr_buffer = Eio.Buf_read.of_flow read_stderr ~max_size:max_int in
   let*? url = get_ws_url ~from:stderr_buffer process in
-  let*? () = Websocket.connect ~sw ~env url in
+  let () = Websocket.connect ~sw ~env url in
   let*? result =
     let open Cdp.Commands.Target.CreateBrowserContext in
     Request.make ?sessionId:None ~params:(Params.make ())

lib/OSnap_Browser/dune

@@ -5,7 +5,8 @@
   OSnap_Websocket
   unix
   fmt
-  piaf
+  httpun
+  httpun-eio
   bigstringaf
   cdp
   decompress.de

lib/OSnap_Websocket/OSnap_Websocket.ml

@@ -7,7 +7,6 @@ let id () =
   !id
 ;;
 
-let close_requests = Queue.create ()
 let pending_requests = Queue.create ()
 let sent_requests = Hashtbl.create 10
 let listeners = Hashtbl.create 10
@@ -20,52 +19,42 @@ let call_event_handlers key message =
       Hashtbl.remove events key))
 ;;
 
-let connect ~sw ~env url =
-  let uri = Uri.of_string url in
-  let resource = Uri.path uri in
-  let*? client =
-    Piaf.Client.create env ~sw (Uri.with_scheme uri (Some "http"))
-    |> Result.map_error (fun _ -> `OSnap_CDP_Connection_Failed)
-  in
-  let*? wsd =
-    Piaf.Client.ws_upgrade client resource
-    |> Result.map_error (fun _ -> `OSnap_CDP_Connection_Failed)
+let websocket_handler ~sw u wsd =
+  let send_payload payload =
+    let payload = Bytes.of_string payload in
+    let len = Bytes.length payload in
+    Httpun_ws.Wsd.send_bytes wsd ~kind:`Text payload ~off:0 ~len
   in
-  let close () = Piaf.Ws.Descriptor.close wsd in
   let rec input_loop () =
     let () = Eio.Fiber.yield () in
-    if not (Queue.is_empty close_requests)
-    then (
-      close_requests |> Queue.iter (fun resolver -> Eio.Promise.resolve resolver ());
-      close_requests |> Queue.clear;
-      close ())
-    else if not (Queue.is_empty pending_requests)
+    if not (Queue.is_empty pending_requests)
     then (
       let key, message, resolver = Queue.take pending_requests in
-      Piaf.Ws.Descriptor.send_string wsd message;
+      let () = send_payload message in
       Hashtbl.add sent_requests key resolver;
       input_loop ())
     else input_loop ()
   in
-  Result.ok
-  @@ Eio.Fiber.fork ~sw
-  @@ fun () ->
-  Eio.Fiber.both
-    (fun () ->
-      input_loop ();
-      Piaf.Client.shutdown client)
-    (fun () ->
-      wsd
-      |> Piaf.Ws.Descriptor.messages
-      |> Piaf.Stream.iter ~f:(fun (_opcode, { Piaf.IOVec.buffer; off; len }) ->
-        let response = Bigstringaf.substring ~off ~len buffer in
-        let json = response |> Yojson.Safe.from_string in
-        let id = json |> Yojson.Safe.Util.member "id" |> Yojson.Safe.Util.to_int_option in
+  let frame ~(opcode : Httpun_ws.Websocket.Opcode.t) ~is_fin:_ ~len payload =
+    match opcode with
+    | `Connection_close | `Continuation | `Other _ -> ()
+    | `Ping -> Httpun_ws.Wsd.send_pong wsd
+    | `Pong -> ()
+    | `Text | `Binary ->
+      let buf = Eio.Buf_write.create len in
+      let on_eof () =
+        let response = Eio.Buf_write.serialize_to_string buf in
+        let json = response |> Yojson.Basic.from_string in
+        let id =
+          json |> Yojson.Basic.Util.member "id" |> Yojson.Basic.Util.to_int_option
+        in
         let method_ =
-          json |> Yojson.Safe.Util.member "method" |> Yojson.Safe.Util.to_string_option
+          json |> Yojson.Basic.Util.member "method" |> Yojson.Basic.Util.to_string_option
         in
         let sessionId =
-          json |> Yojson.Safe.Util.member "sessionId" |> Yojson.Safe.Util.to_string_option
+          json
+          |> Yojson.Basic.Util.member "sessionId"
+          |> Yojson.Basic.Util.to_string_option
         in
         (match method_, sessionId with
          | None, None -> ()
@@ -86,7 +75,61 @@ let connect ~sw ~env url =
           Hashtbl.find_opt sent_requests key
           |> Option.iter (fun resolver -> Eio.Promise.resolve resolver response);
           Hashtbl.remove sent_requests key;
-          ()))
+          ()
+      in
+      let rec on_read bs ~off ~len:read_len =
+        let response = Bigstringaf.substring ~off ~len:read_len bs in
+        Eio.Buf_write.string buf response;
+        Eio.Fiber.yield ();
+        if len > read_len
+        then Httpun_ws.Payload.schedule_read payload ~on_read ~on_eof
+        else on_eof ()
+      in
+      Httpun_ws.Payload.schedule_read payload ~on_read ~on_eof
+  in
+  Eio.Fiber.fork_daemon ~sw input_loop;
+  let eof () = Eio.Promise.resolve u () in
+  Httpun_ws.Websocket_connection.{ frame; eof }
+;;
+
+let error_handler = function
+  | `Handshake_failure (rsp, _body) ->
+    Format.eprintf "Handshake failure: %a\n%!" Httpun.Response.pp_hum rsp
+  | _ -> assert false
+;;
+
+let connect ~sw ~env url =
+  let net = Eio.Stdenv.net env in
+  Eio.Fiber.fork_daemon ~sw (fun () ->
+    let uri = Uri.of_string url in
+    let host = Uri.host_with_default uri ~default:"localhost" in
+    let port = Uri.port uri |> Option.value ~default:80 in
+    let addresses =
+      Eio.Net.getaddrinfo net ~service:"http" host
+      |> List.filter_map
+         @@ function
+         | `Tcp (addr, _port) ->
+           let addr = Eio.Net.Ipaddr.fold ~v4:Option.some ~v6:(Fun.const None) addr in
+           Option.map (fun addr -> `Tcp (addr, port)) addr
+         | _ -> None
+    in
+    let socket = Eio.Net.connect ~sw net (List.hd addresses) in
+    let p, u = Eio.Promise.create () in
+    let nonce = "0123456789ABCDEF" in
+    let resource = Uri.path_and_query uri in
+    let _client =
+      Httpun_ws_eio.Client.connect
+        socket
+        ~sw
+        ~nonce
+        ~host
+        ~port
+        ~resource
+        ~error_handler
+        ~websocket_handler:(websocket_handler ~sw u)
+    in
+    Eio.Promise.await p;
+    `Stop_daemon)
 ;;
 
 let send message =
@@ -111,9 +154,3 @@ let listen ?(look_behind = true) ~event ~sessionId handler =
         Hashtbl.remove listeners key;
         Hashtbl.remove events key))
 ;;
-
-let close () =
-  let p, resolver = Eio.Promise.create () in
-  close_requests |> Queue.add resolver;
-  Eio.Promise.await p
-;;

lib/OSnap_Websocket/OSnap_Websocket.mli

@@ -5,11 +5,5 @@ val listen
   -> (string -> (unit -> unit) -> unit)
   -> unit
 
-val close : unit -> unit
 val send : (int -> string) -> string
-
-val connect
-  :  sw:Eio.Switch.t
-  -> env:Eio_unix.Stdenv.base
-  -> string
-  -> (unit, [> `OSnap_CDP_Connection_Failed ]) result
+val connect : sw:Eio.Switch.t -> env:Eio_unix.Stdenv.base -> string -> unit

lib/OSnap_Websocket/dune

@@ -2,8 +2,6 @@
  (name OSnap_Websocket)
  (libraries
   OSnap_Utils
-  piaf
-  piaf.stream
   bigstringaf
   httpun
   httpun-ws

osnap.opam

@@ -21,7 +21,6 @@ depends: [
   "fmt"
   "libspng"
   "odiff-core"
-  "piaf"
   "re"
   "uri"
   "yaml"
@@ -47,7 +46,4 @@ pin-depends: [
   [ "libspng.dev" "git+https://github.com/eWert-Online/esy-libspng.git#opam"]
   [ "cdp.dev" "git+https://github.com/eWert-Online/reason-cdp.git#502346a7ba9a512ce9aaec456561b64464b3782f"]
   [ "odiff-core.dev" "git+https://github.com/dmtrKovalenko/odiff.git#5a9a1976c553b6c57c32c4e9dcf185bbcdaf1fca"]
-  [ "piaf.dev" "git+https://github.com/anmonteiro/piaf.git"]
-  [ "eio-ssl.dev" "git+https://github.com/anmonteiro/eio-ssl.git#0.3.0" ]
-  [ "multipart_form.dev" "git+https://github.com/anmonteiro/multipart_form.git" ]
 ]

osnap.opam.template

@@ -2,7 +2,5 @@ pin-depends: [
   [ "libspng.dev" "git+https://github.com/eWert-Online/esy-libspng.git#opam"]
   [ "cdp.dev" "git+https://github.com/eWert-Online/reason-cdp.git#502346a7ba9a512ce9aaec456561b64464b3782f"]
   [ "odiff-core.dev" "git+https://github.com/dmtrKovalenko/odiff.git#5a9a1976c553b6c57c32c4e9dcf185bbcdaf1fca"]
-  [ "piaf.dev" "git+https://github.com/anmonteiro/piaf.git"]
-  [ "eio-ssl.dev" "git+https://github.com/anmonteiro/eio-ssl.git#0.3.0" ]
-  [ "multipart_form.dev" "git+https://github.com/anmonteiro/multipart_form.git" ]
+  [ "httpun-ws.dev" "git+https://github.com/eWert-Online/httpun-ws.git#fec8721c4eb67ef0c9436677916473ab81a1ebbe"]
 ]