-
Notifications
You must be signed in to change notification settings - Fork 0
/
protocol.py
59 lines (51 loc) · 1.52 KB
/
protocol.py
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
48
49
50
51
52
53
54
55
56
57
58
59
class ProtocolError(Exception):
pass
class Header():
SPLIT = '$_$'.encode()
PUSH = 'PUSH'.encode()
POP = 'POP'.encode()
RESPONSE = 'RESPONSE'.encode()
METHODS = (PUSH, POP, RESPONSE)
def __init__(self, header_frame):
self.qname = None
self.method = None
self.parse_header(header_frame)
def parse_header(self, header_frame):
try:
method, qname = header_frame.split(self.SPLIT, 1)
assert method in self.METHODS, f'error method: {method}'
except:
raise ProtocolError()
self.qname = qname
self.method = method
class Protocol():
SPLIT = '( ̄﹁ ̄)'.encode()
def __init__(self, frame):
self.frame = frame
self.header = None
self.body = None
self.parser_frame()
def parser_frame(self):
try:
header, body = self.frame.split(self.SPLIT, 1)
header = bytes.fromhex(header.decode())
self.header = Header(header)
self.body = body
except:
raise ProtocolError()
@classmethod
def build_request(cls, method, q_name, body):
return cls.SPLIT.join(
(
Header.SPLIT.join((method, q_name)).hex().encode(),
body
)
)
@classmethod
def build_response(cls, q_name, body):
return cls.SPLIT.join(
(
Header.SPLIT.join((Header.RESPONSE, q_name)).hex().encode(),
body
)
)