Change in vdsm[master]: jsonrpc: Optimize stomp encoder\decoder
smizrahi at redhat.com
smizrahi at redhat.com
Mon Nov 17 22:59:31 UTC 2014
Saggi Mizrahi has uploaded a new change for review.
Change subject: jsonrpc: Optimize stomp encoder\decoder
......................................................................
jsonrpc: Optimize stomp encoder\decoder
Change-Id: I341e46463e654ae0a086cd49af18aa336b86cb6e
Signed-off-by: Saggi Mizrahi <smizrahi at redhat.com>
---
M lib/yajsonrpc/stomp.py
1 file changed, 60 insertions(+), 42 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/55/35255/1
diff --git a/lib/yajsonrpc/stomp.py b/lib/yajsonrpc/stomp.py
index 884bff6..1dda6d7 100644
--- a/lib/yajsonrpc/stomp.py
+++ b/lib/yajsonrpc/stomp.py
@@ -14,8 +14,8 @@
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
import logging
+import os
import socket
-import cStringIO
from threading import Timer, Event
from uuid import uuid4
from collections import deque
@@ -27,6 +27,8 @@
_RE_ESCAPE_SEQUENCE = re.compile(r"\\(.)")
+_RE_ENCODE_CHARS = re.compile(r"[\r\n\\:]")
+
_EC_DECODE_MAP = {
r"\\": "\\",
r"r": "\r",
@@ -34,7 +36,12 @@
r"c": ":",
}
-_ESCAPE_CHARS = (('\\\\', '\\'), ('\\r', '\r'), ('\\n', '\n'), ('\\c', ':'))
+_EC_ENCODE_MAP = {
+ ":": "\c",
+ "\\": "\\\\",
+ "\r": "\\r",
+ "\n": "\\n",
+}
class Command:
@@ -70,6 +77,8 @@
class Frame(object):
+ __slots__ = ("headers", "command", "body")
+
def __init__(self, command="", headers=None, body=None):
self.command = command
if headers is None:
@@ -87,15 +96,19 @@
if body is not None:
self.headers["content-length"] = len(body)
- data = self.command + '\n'
- data += '\n'.join(["%s:%s" % (encodeValue(key), encodeValue(value))
- for key, value in self.headers.iteritems()])
- data += '\n\n'
- if body is not None:
- data += body
+ data = [self.command, '\n']
+ for key, value in self.headers.iteritems():
+ data.append(encodeValue(key))
+ data.append(":")
+ data.append(encodeValue(value))
+ data.append("\n")
- data += "\0"
- return data
+ data.append('\n')
+ if body is not None:
+ data.append(body)
+
+ data.append("\0")
+ return ''.join(data)
def __repr__(self):
return "<StompFrame command=%s>" % (repr(self.command))
@@ -111,24 +124,23 @@
raise ValueError("Contains illigal charachter `:`")
try:
- return _RE_ESCAPE_SEQUENCE.sub(
- lambda m: _EC_DECODE_MAP[m.groups()[0]],
- s
+ s = _RE_ESCAPE_SEQUENCE.sub(
+ lambda m: _EC_DECODE_MAP[m.group(0)],
+ s,
)
except KeyError as e:
raise ValueError("Containes invalid escape squence `\\%s`" % e.args[0])
+ return s.decode('utf-8')
+
def encodeValue(s):
- if not isinstance(s, (str, unicode)):
- s = str(s)
- for escaped, raw in _ESCAPE_CHARS:
- s = s.replace(raw, escaped)
-
if isinstance(s, unicode):
s = s.encode('utf-8')
+ elif not isinstance(s, str):
+ s = str(s)
- return s
+ return _RE_ENCODE_CHARS.sub(lambda m: _EC_ENCODE_MAP[m.group(0)], s)
class Parser(object):
@@ -142,22 +154,29 @@
self._STATE_HEADER: self._parse_header,
self._STATE_BODY: self._parse_body}
self._frames = deque()
- self._state = self._STATE_CMD
+ self._change_state(self._STATE_CMD)
self._contentLength = -1
self._flush()
+ def _change_state(self, new_state):
+ self._state = new_state
+ self._state_cb = self._states[new_state]
+
def _flush(self):
- self._buffer = cStringIO.StringIO()
+ self._buffer = ""
+
+ def _write_buffer(self, buff):
+ self._buffer += buff
+
+ def _get_buffer(self):
+ return self._buffer
def _handle_terminator(self, term):
- if term not in self._buffer.getvalue():
+ res, _, rest = self._buffer.partition(term)
+ if not _:
return None
- data = self._buffer.getvalue()
- res, rest = data.split(term, 1)
- self._flush()
- self._buffer.write(rest)
-
+ self._buffer = rest
return res
def _parse_command(self):
@@ -174,7 +193,7 @@
cmd = decodeValue(cmd)
self._tmpFrame = Frame(cmd)
- self._state = self._STATE_HEADER
+ self._change_state(self._STATE_HEADER)
return True
def _parse_header(self):
@@ -182,13 +201,13 @@
if header is None:
return False
- headers = self._tmpFrame.headers
if len(header) > 0 and header[-1] == '\r':
header = header[:-1]
+ headers = self._tmpFrame.headers
if header == "":
self._contentLength = int(headers.get('content-length', -1))
- self._state = self._STATE_BODY
+ self._change_state(self._STATE_BODY)
return True
key, value = header.split(":", 1)
@@ -206,7 +225,7 @@
def _pushFrame(self):
self._frames.append(self._tmpFrame)
- self._state = self._STATE_CMD
+ self._change_state(self._STATE_CMD)
self._tmpFrame = None
self._contentLength = -1
@@ -226,16 +245,16 @@
return True
def _parse_body_length(self):
- buf = self._buffer
+ buf = self._get_buffer()
cl = self._contentLength
- ndata = buf.tell()
+ ndata = len(buf)
if ndata < cl:
return False
remainingBytes = 0
self._flush()
- body = buf.getvalue()
- self._buffer.write(body[cl + 1:])
+ body = buf
+ self._write_buffer(body[cl + 1:])
body = body[:cl]
if remainingBytes == 0:
@@ -249,9 +268,8 @@
return len(self._frames)
def parse(self, data):
- states = self._states
- self._buffer.write(data)
- while states[self._state]():
+ self._write_buffer(data)
+ while self._state_cb():
pass
def popFrame(self):
@@ -347,7 +365,7 @@
def put_subscribe(self, destination, ack=None):
subid = self._aclient.subscribe(self._adisp, destination, ack)
- sub = Subsciption(self, subid, ack)
+ sub = Subscription(self, subid, ack)
self._registerSubscription(sub)
return sub
@@ -378,7 +396,7 @@
class AsyncDispatcher(object):
log = logging.getLogger("stomp.AsyncDispatcher")
- def __init__(self, frameHandler, bufferSize=1024):
+ def __init__(self, frameHandler, bufferSize=4096):
self._frameHandler = frameHandler
self._bufferSize = bufferSize
self._parser = Parser()
@@ -471,7 +489,7 @@
return True
def _milis(self):
- return int(round(time.time() * 1000))
+ return int(os.times()[4] * 1000)
class AsyncClient(object):
@@ -552,7 +570,7 @@
return subscriptionID
-class Subsciption(object):
+class Subscription(object):
def __init__(self, client, subid, ack):
self._ack = ack
self._subid = subid
--
To view, visit http://gerrit.ovirt.org/35255
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I341e46463e654ae0a086cd49af18aa336b86cb6e
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Saggi Mizrahi <smizrahi at redhat.com>
More information about the vdsm-patches
mailing list