Change in vdsm[master]: jsonrpc: Remove JsonRpcClientPool
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: Remove JsonRpcClientPool
......................................................................
jsonrpc: Remove JsonRpcClientPool
It was created before the reactor was able to manage
client connections as well.
Change-Id: Ib6b67ee2ccb3acfac17a538ad2177f53c0007656
Signed-off-by: Saggi Mizrahi <smizrahi at redhat.com>
---
M lib/yajsonrpc/__init__.py
M tests/jsonRpcHelper.py
2 files changed, 82 insertions(+), 106 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/53/35253/1
diff --git a/lib/yajsonrpc/__init__.py b/lib/yajsonrpc/__init__.py
index b3fd590..69be3c4 100644
--- a/lib/yajsonrpc/__init__.py
+++ b/lib/yajsonrpc/__init__.py
@@ -259,96 +259,6 @@
self.sendReply()
-class JsonRpcClientPool(object):
- def __init__(self, reactor):
- self.log = logging.getLogger("JsonRpcClientPool")
- self._reactor = reactor
- self._inbox = Queue()
- self._clients = {}
- self._eventcbs = []
-
- def createClient(self, connected_socket):
- transport = self._reactor.createClient(connected_socket)
- transport.setMessageHandler(self._handleClientMessage)
- client = JsonRpcClient(transport)
- self._clients[transport] = client
- return client
-
- def _handleClientMessage(self, req):
- self._inbox.put_nowait(req)
-
- def registerEventCallback(self, eventcb):
- self._eventcbs.append(ref(eventcb))
-
- def unregisterEventCallback(self, eventcb):
- for r in self._eventcbs[:]:
- cb = r()
- if cb is None or cb == eventcb:
- try:
- self._eventcbs.remove(r)
- except ValueError:
- # Double unregister, ignore.
- pass
-
- def emit(self, client, event, params):
- for r in self._eventcbs[:]:
- cb = r()
- if cb is None:
- continue
-
- cb(client, event, params)
-
- def _processEvent(self, client, obj):
- if isinstance(obj, list):
- map(self._processEvent, obj)
- return
-
- req = JsonRpcRequest.fromRawObject(obj)
- if not req.isNotification():
- self.log.warning("Recieved non notification, ignoring")
-
- self.emit(client, req.method, req.params)
-
- def serve(self):
- while True:
- data = self._inbox.get()
- if data is None:
- return
-
- transport, message = data
- client = self._clients[transport]
- try:
- mobj = json.loads(message)
- isResponse = self._isResponse(mobj)
- except:
- self.log.exception("Problem parsing message from client")
- transport.close()
- del self._clients[transport]
- continue
-
- if isResponse:
- client._processIncomingResponse(mobj)
- else:
- self._processEvent(client, mobj)
-
- def _isResponse(self, obj):
- if isinstance(obj, list):
- v = None
- for res in map(self._isResponse, obj):
- if v is None:
- v = res
-
- if v != res:
- raise TypeError("batch is mixed")
-
- return v
- else:
- return ("result" in obj or "error" in obj)
-
- def close(self):
- self._inbox.put(None)
-
-
class JsonRpcCall(object):
def __init__(self):
self._ev = Event()
@@ -372,9 +282,11 @@
class JsonRpcClient(object):
def __init__(self, transport):
self.log = logging.getLogger("jsonrpc.JsonRpcClient")
+ transport.setMessageHandler(self._handleMessage)
self._transport = transport
self._runningRequests = {}
self._lock = Lock()
+ self._eventcbs = []
def setTimeout(self, timeout):
self._transport.setTimeout(timeout)
@@ -414,20 +326,20 @@
raise ValueError("Request id already in use %s", rid)
self._runningRequests[rid] = ctx
- self._transport.send(ctx.encode())
+
+ self._transport.send(ctx.encode())
# All notifications
if ctx.isDone():
self._finalizeCtx(ctx)
def _finalizeCtx(self, ctx):
- with self._lock:
- if not ctx.isDone():
- return
+ if not ctx.isDone():
+ return
- cb = ctx.callback
- if cb is not None:
- cb(self, ctx.getResponses())
+ cb = ctx.callback
+ if cb is not None:
+ cb(self, ctx.getResponses())
def _processIncomingResponse(self, resp):
if isinstance(resp, list):
@@ -437,13 +349,73 @@
resp = JsonRpcResponse.fromRawObject(resp)
with self._lock:
ctx = self._runningRequests.pop(resp.id)
- ctx.addResponse(resp)
+
+ ctx.addResponse(resp)
self._finalizeCtx(ctx)
+ def _isResponse(self, obj):
+ if isinstance(obj, list):
+ v = None
+ for res in map(self._isResponse, obj):
+ if v is None:
+ v = res
+
+ if v != res:
+ raise TypeError("batch is mixed")
+
+ return v
+ else:
+ return ("result" in obj or "error" in obj)
+
+ def _handleMessage(self, req):
+ transport, message = req
+ try:
+ mobj = json.loads(message)
+ isResponse = self._isResponse(mobj)
+ except:
+ self.log.exception("Problem parsing message from client")
+
+ if isResponse:
+ self._processIncomingResponse(mobj)
+ else:
+ self._processEvent(mobj)
+
+ def _processEvent(self, obj):
+ if isinstance(obj, list):
+ map(self._processEvent, obj)
+ return
+
+ req = JsonRpcRequest.fromRawObject(obj)
+ if not req.isNotification():
+ self.log.warning("Recieved non notification, ignoring")
+
+ self.emit(req.method, req.params)
+
def close(self):
self._transport.close()
+ def registerEventCallback(self, eventcb):
+ self._eventcbs.append(ref(eventcb))
+
+ def unregisterEventCallback(self, eventcb):
+ for r in self._eventcbs[:]:
+ cb = r()
+ if cb is None or cb == eventcb:
+ try:
+ self._eventcbs.remove(r)
+ except ValueError:
+ # Double unregister, ignore.
+ pass
+
+ def emit(self, event, params):
+ for r in self._eventcbs[:]:
+ cb = r()
+ if cb is None:
+ continue
+
+ cb(self, event, params)
+
class JsonRpcServer(object):
log = logging.getLogger("jsonrpc.JsonRpcServer")
diff --git a/tests/jsonRpcHelper.py b/tests/jsonRpcHelper.py
index 066eaeb..ed7c345 100644
--- a/tests/jsonRpcHelper.py
+++ b/tests/jsonRpcHelper.py
@@ -30,7 +30,7 @@
from rpc.BindingXMLRPC import BindingXMLRPC, XmlDetector
from yajsonrpc.stompReactor import StompDetector
from protocoldetector import MultiProtocolAcceptor
-from yajsonrpc import JsonRpcClientPool
+from yajsonrpc import JsonRpcClient
from rpc.BindingJsonRpc import BindingJsonRpc
from sslhelper import DEAFAULT_SSL_CONTEXT
@@ -105,15 +105,19 @@
reactor = handler._reactor
if not client:
- cpool = JsonRpcClientPool(reactor)
- t = threading.Thread(target=cpool.serve)
- t.setDaemon(True)
- t.start()
- client = cpool.createClient
+ client = lambda client_socket: (
+ JsonRpcClient(reactor.createClient(client_socket))
+ )
_, port = acceptor._socket.getsockname()
- clientFactory = partial(client, create_socket(sslctx, acceptor._host,
- port))
+ clientFactory = partial(
+ client,
+ create_socket(
+ sslctx,
+ acceptor._host,
+ port
+ )
+ )
yield clientFactory
--
To view, visit http://gerrit.ovirt.org/35253
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ib6b67ee2ccb3acfac17a538ad2177f53c0007656
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