blob: 7b103aba0c33a62a4feb70355b2a59f7c77449fe (
plain)
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
|
"""Common JSONRPC framing helpers for jsonrpc.el test servers."""
import json
import sys
def read_msg():
"""Read one Content-Length-framed JSON-RPC message from stdin."""
headers = {}
while True:
line = sys.stdin.buffer.readline()
if not line:
return None
text = line.decode('utf-8').rstrip('\r\n')
if not text:
break
if ':' in text:
k, _, v = text.partition(':')
headers[k.strip()] = v.strip()
n = int(headers.get('Content-Length', 0))
return json.loads(sys.stdin.buffer.read(n).decode('utf-8')) if n else None
def write_msg(msg):
"""Write one Content-Length-framed JSON-RPC message to stdout."""
body = json.dumps(msg, ensure_ascii=False).encode('utf-8')
sys.stdout.buffer.write(
f'Content-Length: {len(body)}\r\n\r\n'.encode('utf-8') + body
)
sys.stdout.buffer.flush()
def log(text):
"""Write a log line to stderr."""
print(f'[test-server] {text}', file=sys.stderr, flush=True)
def harakiri(msg):
"""Maybe handle harakiri request/notif in msg."""
if msg.get('method') == 'harakiri':
log('-> very clean harakiri')
if (mid := msg.get('id')) is not None:
write_msg({'jsonrpc': '2.0', 'id': mid, 'result': True})
return True
return False
|
