Source file sse.ml

type event = {
  event_type : string;
  data : string;
}

let parse_events lines =
  let current_event = ref "" in
  let current_data = Buffer.create 256 in
  let stream, push = Lwt_stream.create () in
  let emit () =
    if String.length !current_event > 0 || Buffer.length current_data > 0 then begin
      let evt = { event_type = !current_event; data = Buffer.contents current_data } in
      push (Some evt);
      current_event := "";
      Buffer.clear current_data
    end
  in
  Lwt.async (fun () ->
    try%lwt
      let%lwt () =
        Lwt_stream.iter
          (fun line ->
            match line with
            | "" -> emit ()
            | line when String.starts_with ~prefix:":" line -> ()
            | line when String.starts_with ~prefix:"event:" line ->
              let value = String.trim (String.sub line 6 (String.length line - 6)) in
              current_event := value
            | line when String.starts_with ~prefix:"data:" line ->
              let value = String.sub line 5 (String.length line - 5) in
              let value =
                if String.length value > 0 && Char.equal (String.get value 0) ' ' then
                  String.sub value 1 (String.length value - 1)
                else value
              in
              if Buffer.length current_data > 0 then Buffer.add_char current_data '\n';
              Buffer.add_string current_data value
            | _ -> ())
          lines
      in
      emit ();
      push None;
      Lwt.return_unit
    with exn ->
      push None;
      Lwt.fail exn);
  stream