src/sse.bend

ezmcp@0.1.0.0 · 101 lines · 12 definitions

1# ezmcp/src/sse: the data of a text/event-stream body, as an MCP client
2# reads a POST's reply when the server streams it
3# (basic/transports#sending-messages-to-the-server). Only `data:` fields
4# are read: an event's data is its data lines joined by newlines, and the
5# reply is the data of the last event that has any.
6import Base
7
8# while reading: the current event's data so far (if it has any), and the
9# data of the last event that ended
10type Sse is Data:
11  Sse{cur: Maybe<&2, String>, last: String}
12
13# one char of a line: a `\r` that ends the line is dropped
14def unr.at(head: Char, rest: Unit -> String, ends: Bool) -> String:
15  match ends:
16    case True{}:
17      ""
18    case False{}:
19      SCon{head, rest(Unit{})}
20
21# a line's text with one `\r` at its end dropped
22def unr(text: String) -> String:
23  match text:
24    case SNil{}:
25      SNil{}
26    case SCon{+head, +tail}:
27      unr.at(head, _u => unr(tail),
28        Bool.and(Char.is_eq(head, '\r'), String.is_empty(tail)))
29
30# a data field's value: the text after `data:`, less one space
31def value(text: String) -> String:
32  match text:
33    case SNil{}:
34      SNil{}
35    case SCon{+h, +t}:
36      Bool.pick(String, Char.is_eq(h, ' '), t, SCon{h, t})
37
38# more data for the current event
39def more(cur: Maybe<&2, String>, val: String) -> Maybe<&2, String>:
40  match cur:
41    case None{}:
42      Some{val}
43    case Some{d}:
44      Some{d ++ "\n" ++ val}
45
46# the end of an event: its data, if any, is the latest
47def end(cur: Maybe<&2, String>, last: String) -> Sse:
48  match cur:
49    case None{}:
50      Sse{None{}, last}
51    case Some{d}:
52      Sse{None{}, d}
53
54def line.data(
55  cur: Maybe<&2, String>,
56  last: String,
57  +line: String,
58  data: Bool
59) -> Sse:
60  match data:
61    case True{}:
62      Sse{more(cur, value(String.drop(line, 5n))), last}
63    case False{}:
64      Sse{cur, last}
65
66def line.of(
67  cur: Maybe<&2, String>,
68  last: String,
69  +line: String,
70  blank: Bool
71) -> Sse:
72  match blank:
73    case True{}:
74      end(cur, last)
75    case False{}:
76      line.data(cur, last, line, String.starts_with(line, "data:"))
77
78# one line of the stream
79def line(text: Sse, +line2: String) -> Sse:
80  match text:
81    case Sse{cur, last}:
82      line.of(cur, last, line2, String.is_empty(line2))
83
84# every line of the stream, in order
85def lines(ls: List<&2, String>, text: Sse) -> Sse:
86  match ls:
87    case []:
88      text
89    case l <> rest:
90      lines(rest, line(text, unr(l)))
91
92# the data the stream ends with: an unfinished event's, else the last one's
93def fin(text: Sse) -> String:
94  match text:
95    case Sse{cur, last}:
96      Maybe.default(&2, String, cur, last)
97
98# the data of the last event that has any; an event the stream cut off
99# before its blank line still counts
100def last(body: String) -> String:
101  fin(lines(String.lines(body), Sse{None{}, ""}))