open Lwt.Syntax open Lwt.Infix open Js_of_ocaml open Markup let jss = Js.string let sjs = Js.to_string type t = (signal, async) stream * ((signal, sync) stream option -> unit) let xmlns = "urn:ietf:params:xml:ns:xmpp-framing" (* sic. XEP-0156: "host-meta files MUST be fetched only over HTTPS". I don't make the rules. *) let well_known_of (domain : string) = "https://" ^ domain ^ "/.well-known/host-meta" (** [open_stanza domain] is an stanza for [domain]. If [from] is specified, the stanza has the from parameter. *) let header ?from domain (stream, push) = let stanza = let attributes = let open Option in [(("", "xmlns"), xmlns); (("", "to"), domain)] @ (map (fun jid -> (("", "from"), jid)) from |> to_list) @ [(("", "version"), "1.0")] in [`Start_element ((xmlns, "open"), attributes); `End_element] in push (Some (of_list stanza)); let some_id ((_, name), value) = if name = "id" then Some value else None in let* stanza_open = Markup_lwt.next stream in let* id = match stanza_open with | Some `Start_element ((ns, "open"), attributes) when ns = xmlns -> let* close = Markup_lwt.next stream in begin match close with | Some `End_element -> List.find_map some_id attributes |> Lwt.return | _ -> Lwt.return_none end | _ -> Lwt.return_none in match id with | Some id -> Lwt.return id | None -> Lwt.fail_with "Invalid stream opening server-side." let close (_, p) = {||} |> string |> parse_xml |> signals |> Option.some |> p exception MalformedStanza of Markup.location * Markup.Error.t (** [ws_endpoint domain] is a promise containing the XMPP websocket endpoint associated with [domain], by using the domain's Web-host Metadata. This function uses XMLHttpRequest, so while it should work fine in the browser, in environments that don't provide this constructor (Node.js), there should be some sort of polyfill. Lastly, if [domain] doesn't provide a well-formed Web-host Metadata file, the function throws an exception. *) let ws_endpoint (domain : string) : string Lwt.t = let open Markup in (* This ugly function extracts the href element from a Link tag's attributes if it's a websocket. *) let link_websocket = function | ((_, "rel"), "urn:xmpp:alt-connections:websocket") :: ((_, "href"), href) :: _ | ((_, "href"), href) :: ((_, "rel"), "urn:xmpp:alt-connections:websocket") :: _ -> Some href | _ -> None in let parse_xrd xrd = string xrd |> parse_xml |> signals |> tree ~element:(fun (_, name) attributes children -> match name with | "Link" -> link_websocket attributes | "XRD" -> List.find_map (fun x -> x) children | _ -> None ) |> Option.join in let* host_meta = Js_of_ocaml_lwt.XmlHttpRequest.perform_raw_url (well_known_of domain) in match parse_xrd host_meta.content with | Some x -> Lwt.return x | None -> Lwt.fail_with (domain ^ ": no WebSocket endpoint in Web-host Metadata.") (** [ws_stream url] is a promise to a framed Lwt stream (and its push function) communicating with the websocket located at [url] using the XMPP protocol. Valid XMPP WebSocket subprotocol frames must be sent to the stream, because it directly exposes the websocket under. Pushing [None] closes the websocket. If the websocket is closed server-side, the stream closes. *) let ws_stream (url : string) = let open Lwt_stream in let stream, message = create () in let open Lwt_condition in let is_open = create () in let (ws : WebSockets.webSocket Js.t) = new%js WebSockets.webSocket_withProtocol (jss url) (jss "xmpp") in let push = function | Some msg -> ws##send (jss msg) | None -> ws##close in ws##.onclose := Dom.handler (fun _ -> message None; Js._true); ws##.onmessage := Dom.handler (fun x -> Some (sjs x##.data) |> message; Js._false); ws##.onopen := Dom.handler (fun _ -> signal is_open (); Js._false); let+ () = wait is_open in stream, push (** [connect domain] is an Lwt stream (and its push function) communicating with the XMPP server running at [domain] via the Websocket subprotocol. This function is a complex wrapper around ws_stream, that accepts Markup.ml signals and sends framed XML stanzas to the underlying socket, with exactly one stanza per frame, according to RFC 7935. In essence, it (tries to) expose an identical interface to the original XMPP streamed protocol. Here's an ASCII rendered flow of the data through the various streams. / -> push -> mu_stream -> to_frames -> ws_push -> \ function user websocket \ <--- stream <--- markup_lwt <--- ws_stream <--- / *) let connect domain = let+ ws_stream, ws_push = ws_endpoint domain >>= ws_stream in let open Markup_lwt in (* When sending a malformed stanza (one that Markup.ml doesn't like), a MalformedStanza exception is raised. *) let report loc err = raise (MalformedStanza (loc, err)) in (* Consumes a stream of Markup.ml signals into a series of frames sent to the WebSocket. *) let stanza = ref [] in let total_depth = ref 0 in let stanza_to_string stanza = Markup.(!stanza |> List.rev |> of_list |> write_xml |> to_string) in (* Consume a single fragment, and add it to the "stanza" ref if it's not complete -- as soon as it's completed, send it. *) let chomp_fragment depth fragment = let depth = match fragment with | `Start_element _ -> depth + 1 | `End_element -> depth - 1 | _ -> depth in stanza := fragment :: !stanza; total_depth := depth; (if depth = 0 then let s = [stanza_to_string stanza] in stanza := []; s else []), Some depth in let push = function | None -> ws_push None | Some fragment -> Lwt.async (fun () -> content fragment |> Markup.transform chomp_fragment !total_depth |> Markup_lwt.to_lwt_stream |> Lwt_stream.iter (fun s -> ws_push (Some s))) in ws_stream |> lwt_stream |> strings_to_bytes |> parse_xml ~report |> signals, push