feat!(portal_ws): to_frames function relatively complete

1 files changed, 28 insertions, 4 deletions

diff --git a/portal/lib/portal_ws.ml b/portal/lib/portal_ws.ml
index cd9c08f..ac2f0de 100644
--- a/portal/lib/portal_ws.ml
+++ b/portal/lib/portal_ws.ml
@@ -91,11 +91,35 @@ let connect domain =
   let+ stream, ws_push = ws_endpoint domain >|= ws_stream in
   let streamed_stanzas, push = create () in
   (* Consumes a stream of stanzas fragments into a series of frames sent to the WebSocket. *)
-  (* Right now... This doesn't do much. *)
   let to_frames fragments =
     let stanzas =
-      let open Markup_lwt in
-      lwt_stream fragments |> to_lwt_stream
-    in let+ _ = Lwt_stream.iter (fun x -> ws_push (Some x)) stanzas in ws_push None
+      (* Convert a stream of strings to a stream of characters. *)
+      let spliced =
+        fragments
+        |> map (fun fragment -> String.to_seq fragment |> of_seq)
+        |> concat
+      in
+      let open Markup in
+      spliced
+      |> Markup_lwt.lwt_stream
+      |> Markup_lwt.parse_xml
+      |> signals
+    (*  XML declarations are not to be transmitted to the underlying WebSocket,
+        per IETF recommendation. https://datatracker.ietf.org/doc/html/rfc7395#section-3.3.3 *)
+      |> content
+      |> transform
+           (* Parse well-formed signals. As soon as we have enough elements to send a well-formed signal, we send it to
+              the underlying WebSocket. Currently, sending a non-well-formed signal is... undefined behavior? *)
+           (fun (depth, signals) s ->
+             let depth = match s with
+               | `Start_element _ -> depth + 1
+               | `End_element -> depth - 1
+               | _ -> depth
+             in if depth = 0
+                then [List.rev (s :: signals) |> of_list |> write_xml |> to_string], Some (0, [])
+                else [], Some (depth, s :: signals))
+           (0, [])
+      |> Markup_lwt.to_lwt_stream
+    in let+ _ = iter (fun x -> ws_push (Some x)) stanzas in ws_push None
   in Lwt.async @@ (fun () -> to_frames streamed_stanzas);
      stream, push