-
Notifications
You must be signed in to change notification settings - Fork 0
/
protocodecs.py
112 lines (90 loc) · 2.82 KB
/
protocodecs.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
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
import json
class Request:
def __init__(self, method, key, value):
self.method = method
self.key = key
self.value = value
class Response:
def __init__(self, value, error):
self.value = value
self.error = error
def is_hit(self):
return (self.value is not None) and (self.error is False)
def get_value(self):
return self.value
def is_error(self):
return self.error
def is_ack(self):
return (self.value is None) and (self.error is False)
"""
A very simple cache protocol.
Keys and values are interpreted as utf-8 strings and keys must be
alphanumerical.
Requests
========
Set: {"m": "SET", "k": <key>, "v": <value>}
Get: {"m": "GET", "k": <key>}
Evict: {"m": "EVICT", "k": <key>}
Clear: {"m": "CLEAR"}
Responses
=========
Hit (Get): {"v": <value>, "e": False}
Error (Get, Evict): {"v": None, "e": True}
Ack (Set, Evict, Clear): {"v": None, "e": False}
"""
class SimpleJsonCodec:
def _check_key(self, key):
if not (type(key) is str):
raise TypeError("Key must be of type str.")
if not key.isalnum():
raise ValueError('Key must be alphanumeric.')
def _encode_request(self, method, key=None, value=None, check_key=True):
request = {"m": method}
if key:
self._check_key(key)
request["k"] = key
if value:
request["v"] = value
return bytes(json.dumps(request), 'utf-8')
""" Used by client """
def encode_set(self, key, value):
self._check_key(key)
if not (type(value) is str):
raise TypeError("Value must be of type str.")
return self._encode_request('SET', key, value)
""" Used by client """
def encode_get(self, key):
self._check_key(key)
return self._encode_request('GET', key)
""" Used by client """
def encode_evict(self, key):
self._check_key(key)
return self._encode_request('EVICT', key)
""" Used by client """
def encode_clear(self):
return self._encode_request('CLEAR')
""" Used by client """
def decode_response(self, data):
data = data.decode('utf-8')
j = json.loads(data)
response = Response(j["v"], j["e"])
return response
""" Used by server """
def decode_request(self, data):
data = data.decode('utf-8')
j = json.loads(data)
if j["m"] != "CLEAR":
self._check_key(j["k"])
if not "k" in j:
key = None
else:
key = j["k"]
if not "v" in j:
value = None
else:
value = j["v"]
request = Request(j["m"], key, value)
return request
""" Used by server """
def encode_response(self, value, error):
return bytes(json.dumps({'v': value, 'e': error}), 'utf-8')