Coverage for vibefpmgui/wire.py: 98%

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): 49 ↛ 33line 49 didn't jump to line 33 because the condition on line 49 was always true

50 messages.append(message)