lib/stream.ml


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

open Lwt.Syntax
open Markup

exception ClosedStream
exception InvalidStanza of string

let get (stream : (signal, async) stream) : (signal, sync) stream Lwt.t =
  (** [stanza stream] is a promise containing a full stanza of the fragments of [stream].*)

  let traverse_stanza depth fragment =
    let depth = match fragment with
      | `Start_element _ -> depth + 1
      | `End_element -> depth - 1
      | _ -> depth
    in ([fragment], if depth = 0 then None else Some depth)
  in transform traverse_stanza 0 stream |> Markup_lwt.load

let start domain : Portal.t Lwt.t =
  (** [start domain] is a promise containing a Portal (stream * push) connected to the XMPP server [domain].

      Currently, it doesn't handle anything except the initial [<open/>] stanza. *)
  let* stream, _push = Portal.connect domain
  in let push = function
       | None -> _push (Some Portal.stanza_close);
                 _push None;
       | anything -> _push anything
     in Some (Portal.stanza_open domain) |> push;
        (* TODO: check this is a good stanza *)
        let+ _ = get stream
        in stream, push