vibefpmgui.wire

The framing the Language Server Protocol and the Debug Adapter Protocol share: a Content-Length header, a blank line, then a JSON message (see vibefpmgui.lsp and vibefpmgui.dap).

 1"""The framing the Language Server Protocol and the Debug Adapter Protocol
 2share: a ``Content-Length`` header, a blank line, then a JSON message (see
 3`vibefpmgui.lsp` and `vibefpmgui.dap`)."""
 4
 5from __future__ import annotations
 6
 7import json
 8import re
 9
10_LENGTH = re.compile(rb"Content-Length:\s*(\d+)", re.IGNORECASE)
11
12
13def frame(message: dict) -> bytes:
14    """A message, ready to write."""
15    body = json.dumps(message).encode("utf-8")
16    return b"Content-Length: %d\r\n\r\n" % len(body) + body
17
18
19class FrameReader:
20    """The messages in a stream, read as it arrives (a message may come in
21    pieces, or several at once). Something that isn't a header is skipped,
22    and a message that isn't JSON dropped."""
23
24    def __init__(self):
25        self.buffer = b""
26
27    def clear(self):
28        self.buffer = b""
29
30    def feed(self, data: bytes) -> list[dict]:
31        self.buffer += data
32        messages = []
33        while True:
34            end = self.buffer.find(b"\r\n\r\n")
35            if end < 0:
36                return messages
37            match = _LENGTH.search(self.buffer[:end])
38            if not match:  # not a header: skip it
39                self.buffer = self.buffer[end + 4:]
40                continue
41            length = int(match.group(1))
42            if len(self.buffer) < end + 4 + length:
43                return messages
44            body, self.buffer = self.buffer[end + 4:end + 4 + length], self.buffer[end + 4 + length:]
45            try:
46                message = json.loads(body)
47            except ValueError:
48                continue
49            if isinstance(message, dict):
50                messages.append(message)
def frame(message: dict) -> bytes: 
14def frame(message: dict) -> bytes:
15    """A message, ready to write."""
16    body = json.dumps(message).encode("utf-8")
17    return b"Content-Length: %d\r\n\r\n" % len(body) + body

A message, ready to write.

class FrameReader: 
20class FrameReader:
21    """The messages in a stream, read as it arrives (a message may come in
22    pieces, or several at once). Something that isn't a header is skipped,
23    and a message that isn't JSON dropped."""
24
25    def __init__(self):
26        self.buffer = b""
27
28    def clear(self):
29        self.buffer = b""
30
31    def feed(self, data: bytes) -> list[dict]:
32        self.buffer += data
33        messages = []
34        while True:
35            end = self.buffer.find(b"\r\n\r\n")
36            if end < 0:
37                return messages
38            match = _LENGTH.search(self.buffer[:end])
39            if not match:  # not a header: skip it
40                self.buffer = self.buffer[end + 4:]
41                continue
42            length = int(match.group(1))
43            if len(self.buffer) < end + 4 + length:
44                return messages
45            body, self.buffer = self.buffer[end + 4:end + 4 + length], self.buffer[end + 4 + length:]
46            try:
47                message = json.loads(body)
48            except ValueError:
49                continue
50            if isinstance(message, dict):
51                messages.append(message)

The messages in a stream, read as it arrives (a message may come in pieces, or several at once). Something that isn't a header is skipped, and a message that isn't JSON dropped.

buffer
def clear(self): 
28    def clear(self):
29        self.buffer = b""
def feed(self, data: bytes) -> list[dict]: 
31    def feed(self, data: bytes) -> list[dict]:
32        self.buffer += data
33        messages = []
34        while True:
35            end = self.buffer.find(b"\r\n\r\n")
36            if end < 0:
37                return messages
38            match = _LENGTH.search(self.buffer[:end])
39            if not match:  # not a header: skip it
40                self.buffer = self.buffer[end + 4:]
41                continue
42            length = int(match.group(1))
43            if len(self.buffer) < end + 4 + length:
44                return messages
45            body, self.buffer = self.buffer[end + 4:end + 4 + length], self.buffer[end + 4 + length:]
46            try:
47                message = json.loads(body)
48            except ValueError:
49                continue
50            if isinstance(message, dict):
51                messages.append(message)