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
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
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