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