Change in vdsm[master]: jsonrpc: Introduce the definitive JsonRpcClient

smizrahi at redhat.com smizrahi at redhat.com
Thu Feb 21 21:37:38 UTC 2013


Saggi Mizrahi has uploaded a new change for review.

Change subject: jsonrpc: Introduce the definitive JsonRpcClient
......................................................................

jsonrpc: Introduce the definitive JsonRpcClient

Supports events batch calls and what-not

If only I actually had tests for it

Change-Id: Ie514c9029416a5869802f065b49f2da033ff3da6
Signed-off-by: Saggi Mizrahi <smizrahi at redhat.com>
---
M tests/jsonRpcTests.py
M tests/jsonRpcUtils.py
M vdsm.spec.in
M yajsonrpc/Makefile.am
M yajsonrpc/__init__.py
M yajsonrpc/asyncoreReactor.py
D yajsonrpc/client.py
M yajsonrpc/protonReactor.py
8 files changed, 402 insertions(+), 186 deletions(-)


  git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/04/12304/1

diff --git a/tests/jsonRpcTests.py b/tests/jsonRpcTests.py
index e13e24b..1697eaa 100644
--- a/tests/jsonRpcTests.py
+++ b/tests/jsonRpcTests.py
@@ -36,7 +36,8 @@
 from yajsonrpc import \
     JsonRpcError, \
     JsonRpcMethodNotFoundError, \
-    JsonRpcInternalError
+    JsonRpcInternalError, \
+    JsonRpcRequest
 
 
 CALL_TIMEOUT = 5
@@ -60,6 +61,7 @@
         while True:
             try:
                 client, msg = self._queue.get()
+                self.log.info("Echoing message A")
                 if client is None:
                     return
 
@@ -67,6 +69,22 @@
                 client.send(msg)
             except Exception:
                 self.log.error("EchoServer died unexpectedly", exc_info=True)
+
+
+class ReactorClientSyncWrapper(object):
+    def __init__(self, client):
+        self._client = client
+        self._queue = Queue()
+        self._client.setInbox(self._queue)
+
+    def send(self, data):
+        self._client.send(data)
+
+    def connect(self):
+        self._client.connect()
+
+    def recv(self, timeout=None):
+        return self._queue.get(True, timeout)[1]
 
 
 @expandPermutations
@@ -88,7 +106,7 @@
                 self.fail("Reactor died: (%s) %s" % (type(e), e))
 
         with constructReactor(reactorType) as \
-                (reactor, clientFactory, laddr):
+                (reactor, clientReactor, laddr):
 
             t = threading.Thread(target=echosrv.serve)
             t.setDaemon(True)
@@ -96,14 +114,15 @@
 
             reactor.createListener(laddr, echosrv.accept)
 
-            clientNum = 4
-            repeats = 2
-            subRepeats = 4
+            clientNum = 2
+            repeats = 10
+            subRepeats = 10
 
             clients = []
             try:
                 for i in range(clientNum):
-                    c = clientFactory()
+                    c = ReactorClientSyncWrapper(
+                        clientReactor.createClient(laddr))
                     c.connect()
                     clients.append(c)
 
@@ -111,19 +130,21 @@
                     for client in clients:
                         for i in range(subRepeats):
                             self.log.info("Sending message...")
-                            client.send(data, CALL_TIMEOUT)
+                            client.send(data)
 
                 for i in range(repeats * subRepeats):
                     for client in clients:
-                            self.log.info("Waiting for reply...")
-                            retData = client.recv(CALL_TIMEOUT)
-                            self.log.info("Asserting reply...")
-                            self.assertEquals(
-                                retData, data,
-                                "Data is not as expected " +
-                                "'%s...%s' != '%s...%s'" %
-                                (retData[:10], retData[-10:],
-                                 data[:10], data[-10:]))
+                        self.log.info("Waiting for reply...")
+                        retData = client.recv(CALL_TIMEOUT)
+                        self.log.info("Asserting reply...")
+                        self.assertTrue(isinstance(retData,
+                                                   (str, unicode)))
+                        self.assertEquals(
+                            retData, data,
+                            "Data is not as expected " +
+                            "'%s...%s' != '%s...%s'" %
+                            (retData[:10], retData[-10:],
+                             data[:10], data[-10:]))
             finally:
                 queue.put((None, None))
 
@@ -138,6 +159,17 @@
 
 @expandPermutations
 class JsonRpcServerTests(TestCaseBase):
+
+    def _callTimeout(self, client, methodName, params=None, rid=None,
+                     timeout=None):
+        call = client.call_async(JsonRpcRequest(methodName, params, rid))
+        self.assertTrue(call.wait(timeout))
+        resp = call.responses[0]
+        if resp.error is not None:
+            raise JsonRpcError(resp.error['code'], resp.error['message'])
+
+        return resp.result
+
     @permutations(REACTOR_TYPE_PERMUTATIONS)
     def testMethodCallArgList(self, reactorType):
         data = dummyTextGenerator(1024)
@@ -147,9 +179,8 @@
             client = clientFactory()
             client.connect()
             with closing(client):
-                self.assertEquals(client.callMethod("echo", (data,), 10,
-                                                    CALL_TIMEOUT),
-                                  data)
+                self.assertEquals(self._callTimeout(client, "echo", (data,),
+                                  10, CALL_TIMEOUT), data)
 
     @permutations(REACTOR_TYPE_PERMUTATIONS)
     def testMethodCallArgDict(self, reactorType):
@@ -160,7 +191,7 @@
             client = clientFactory()
             client.connect()
             with closing(client):
-                self.assertEquals(client.callMethod("echo",
+                self.assertEquals(self._callTimeout(client, "echo",
                                                     {'text': data},
                                                     10, CALL_TIMEOUT),
                                   data)
@@ -173,7 +204,7 @@
             client.connect()
             with closing(client):
                 with self.assertRaises(JsonRpcError) as cm:
-                    client.callMethod("I.DO.NOT.EXIST :(", [], 10,
+                    self._callTimeout(client, "I.DO.NOT.EXIST :(", [], 10,
                                       CALL_TIMEOUT)
 
                 self.assertEquals(cm.exception.code,
@@ -189,7 +220,8 @@
             client.connect()
             with closing(client):
                 with self.assertRaises(JsonRpcError) as cm:
-                    client.callMethod("echo", [], 10, timeout=CALL_TIMEOUT)
+                    self._callTimeout(client, "echo", [], 10,
+                                      timeout=CALL_TIMEOUT)
 
                 self.assertEquals(cm.exception.code,
                                   JsonRpcInternalError().code)
@@ -201,6 +233,7 @@
             client = clientFactory()
             client.connect()
             with closing(client):
-                res = client.callMethod("ping", [], 10, timeout=CALL_TIMEOUT)
+                res = self._callTimeout(client, "ping", [], 10,
+                                        timeout=CALL_TIMEOUT)
 
                 self.assertEquals(res, None)
diff --git a/tests/jsonRpcUtils.py b/tests/jsonRpcUtils.py
index 99e63d8..43b2ffd 100644
--- a/tests/jsonRpcUtils.py
+++ b/tests/jsonRpcUtils.py
@@ -8,9 +8,8 @@
 
 from yajsonrpc import \
     JsonRpcServer, \
-    asyncoreReactor
-from yajsonrpc.client import \
-    JsonRpcClient
+    asyncoreReactor, \
+    JsonRpcClientPool
 
 protonReactor = None
 try:
@@ -36,21 +35,8 @@
 def _tcpServerConstructor():
     port = getFreePort()
     address = ("localhost", port)
-    reactor = asyncoreReactor.AsyncoreReactor()
-    clientsReactor = asyncoreReactor.AsyncoreReactor()
-
-    t = threading.Thread(target=clientsReactor.process_requests)
-    t.setDaemon(True)
-    t.start()
-
-    def clientFactory(address):
-        return TestClientWrapper(clientsReactor.createClient(address))
-
-    try:
-        yield reactor, partial(clientFactory, address), address
-    finally:
-        reactor.stop()
-        clientsReactor.stop()
+    reactorType = asyncoreReactor.AsyncoreReactor
+    yield reactorType, address
 
 
 @contextmanager
@@ -59,59 +45,43 @@
         raise SkipTest("qpid-proton python bindings are not installed")
 
     port = getFreePort()
-    reactor = protonReactor.ProtonReactor()
+    reactorType = protonReactor.ProtonReactor
 
-    def clientFactory(address):
-        return TestClientWrapper(reactor.createClient(address))
-
-    try:
-        yield (reactor,
-               partial(clientFactory, ("127.0.0.1", port)),
-               ("127.0.0.1", port))
-    finally:
-        reactor.stop()
+    yield (reactorType,
+           ("127.0.0.1", port))
 
 
 REACTOR_CONSTRUCTORS = {"tcp": _tcpServerConstructor,
-                        "proton": _protonServerConstructor}
+                        "amqp": _protonServerConstructor}
 REACTOR_TYPE_PERMUTATIONS = [[r] for r in REACTOR_CONSTRUCTORS.iterkeys()]
 
 
 @contextmanager
 def constructReactor(tp):
-    with REACTOR_CONSTRUCTORS[tp]() as (serverReactor, clientFactory, laddr):
+    with REACTOR_CONSTRUCTORS[tp]() as (reactorType, laddr):
 
+        serverReactor = reactorType()
         t = threading.Thread(target=serverReactor.process_requests)
         t.setDaemon(True)
         t.start()
 
-        yield serverReactor, clientFactory, laddr
+        clientReactor = reactorType()
+        t = threading.Thread(target=clientReactor.process_requests)
+        t.setDaemon(True)
+        t.start()
 
-
-class TestClientWrapper(object):
-    def __init__(self, client):
-        self._client = client
-        self._queue = Queue()
-        self._client.setInbox(self._queue)
-
-    def send(self, data, timeout=None):
-        self._client.send(data)
-
-    def recv(self, timeout=None):
-        return self._queue.get(timeout=timeout)[1]
-
-    def connect(self):
-        return self._client.connect()
-
-    def close(self):
-        return self._client.close()
+        try:
+            yield serverReactor, clientReactor, laddr
+        finally:
+            clientReactor.stop()
+            serverReactor.stop()
 
 
 @contextmanager
 def constructServer(tp, bridge):
     queue = Queue()
     server = JsonRpcServer(bridge, queue)
-    with constructReactor(tp) as (reactor, clientFactory, laddr):
+    with constructReactor(tp) as (reactor, clientReactor, laddr):
         def _accept(listener, client):
             client.setInbox(queue)
 
@@ -121,10 +91,15 @@
         t.setDaemon(True)
         t.start()
 
-        def jsonClientFactory():
-            return JsonRpcClient(clientFactory())
+        cpool = JsonRpcClientPool({tp: clientReactor})
+        t = threading.Thread(target=cpool.serve)
+        t.setDaemon(True)
+        t.start()
+
+        curl = "%s://%s:%d" % (tp, laddr[0], laddr[1])
+        clientFactory = partial(cpool.createClient, curl)
 
         try:
-            yield server, jsonClientFactory
+            yield server, clientFactory
         finally:
             server.stop()
diff --git a/vdsm.spec.in b/vdsm.spec.in
index 38838ea..c3b0296 100644
--- a/vdsm.spec.in
+++ b/vdsm.spec.in
@@ -988,7 +988,6 @@
 %{python_sitearch}/yajsonrpc/__init__.py*
 %{python_sitearch}/yajsonrpc/asyncoreReactor.py*
 %{python_sitearch}/yajsonrpc/protonReactor.py*
-%{python_sitearch}/yajsonrpc/client.py*
 
 %files bootstrap
 %defattr(-, root, root, -)
diff --git a/yajsonrpc/Makefile.am b/yajsonrpc/Makefile.am
index e43eace..422abe9 100644
--- a/yajsonrpc/Makefile.am
+++ b/yajsonrpc/Makefile.am
@@ -21,7 +21,6 @@
 dist_yajsonrpc_PYTHON = \
 	__init__.py \
 	asyncoreReactor.py \
-	client.py \
 	protonReactor.py \
 	$(NULL)
 
diff --git a/yajsonrpc/__init__.py b/yajsonrpc/__init__.py
index 42e0704..65316dd 100644
--- a/yajsonrpc/__init__.py
+++ b/yajsonrpc/__init__.py
@@ -15,6 +15,9 @@
 import json
 import logging
 from functools import partial
+from Queue import Queue
+from weakref import ref
+from threading import Lock, Event
 
 __all__ = ["tcpReactor"]
 
@@ -88,8 +91,6 @@
             raise JsonRpcInvalidRequestError()
 
         reqId = obj.get("id")
-        if not isinstance(reqId, int):
-            raise JsonRpcInvalidRequestError()
 
         params = obj.get('params', [])
         if not isinstance(params, (list, dict)):
@@ -140,6 +141,58 @@
         reqId = obj.get('id')
         return JsonRpcResponse(result, error, reqId)
 
+    @staticmethod
+    def fromRawObject(obj):
+        print obj
+        if obj.get("jsonrpc") != "2.0":
+            raise JsonRpcInvalidRequestError()
+
+        if "result" not in obj and "error" not in obj:
+            raise JsonRpcInvalidRequestError()
+
+        result = obj.get("result")
+        error = obj.get("error")
+
+        reqId = obj.get("id")
+        if not isinstance(reqId, int):
+            raise JsonRpcInvalidRequestError()
+
+        return JsonRpcResponse(result, error, reqId)
+
+
+class _JsonRpcClientRequestContext(object):
+    def __init__(self, requests, callback):
+        self.callback = callback
+        self._requests = requests
+
+        self._responses = {}
+        for req in requests:
+            if req.id is None:
+                continue  # Notifications don't have responses
+
+            self._responses[req.id] = None
+
+    def addResponse(self, resp):
+        self._responses[resp.id] = resp
+
+    def isDone(self):
+        for v in self._responses.values():
+            if v is None:
+                return False
+
+        return True
+
+    def getResponses(self):
+        return self._responses.values()
+
+    def ids(self):
+        return self._responses.keys()
+
+    def encode(self):
+        return ("[" +
+                ", ".join(r.encode() for r in self._requests) +
+                "]")
+
 
 class _JsonRpcServeRequestContext(object):
     def __init__(self, client):
@@ -190,6 +243,191 @@
         self.sendReply()
 
 
+class JsonRpcClientPool(object):
+    log = logging.getLogger("JsonRpcClientPool")
+
+    def __init__(self, reactors):
+        self._reactors = reactors
+        self._inbox = Queue()
+        self._clients = {}
+        self._eventcbs = []
+
+    def createClient(self, address):
+        rtype, address = address.split("://")
+        host, port = address.split(":")
+        port = int(port)
+        transport = self._reactors[rtype].createClient((host, port))
+        transport.setInbox(self._inbox)
+        client = JsonRpcClient(transport)
+        self._clients[transport] = client
+        return client
+
+    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:
+                    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.warning("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()
+        self.responses = None
+
+    def _callback(self, c, resp):
+        if not isinstance(resp, list):
+            resp = [resp]
+
+        self.responses = resp
+        self._ev.set()
+
+    def wait(self, timeout=None):
+        self._ev.wait(timeout)
+        return self._ev.is_set()
+
+    def isSet(self):
+        return self._ev.is_set()
+
+
+class JsonRpcClient(object):
+    log = logging.getLogger("jsonrpc.JsonRpcClient")
+
+    def __init__(self, transport):
+        self._transport = transport
+        self._runningRequests = {}
+        self._lock = Lock()
+
+    def connect(self):
+        self._transport.connect()
+
+    def callMethod(self, methodName, params=[], rid=None):
+        return self.call(JsonRpcRequest(methodName, params, rid))
+
+    def call(self, req):
+        resp = self.call_batch([req])[0]
+        if "error" in resp:
+            raise JsonRpcError(resp.error['code'], resp.error['message'])
+
+        return resp.result
+
+    def call_batch(self, *reqs):
+        call = self.call_async(reqs)
+        call.wait()
+        return call.responses
+
+    def call_async(self, *reqs):
+        call = JsonRpcCall()
+        self.call_cb(call._callback, *reqs)
+        return call
+
+    def call_cb(self, cb, *reqs):
+        ctx = _JsonRpcClientRequestContext(reqs, cb)
+        with self._lock:
+            for rid in ctx.ids():
+                try:
+                    self._runningRequests[rid]
+                except KeyError:
+                    pass
+                else:
+                    raise ValueError("Request id already in use %s", rid)
+
+                self._runningRequests[rid] = ctx
+                self._transport.send(ctx.encode())
+
+        # All notifications
+        if ctx.isDone():
+            self._finalizeCtx(ctx)
+
+    def _finalizeCtx(self, ctx):
+        with self._lock:
+            if not ctx.isDone():
+                print ctx._responses
+                return
+
+            cb = ctx.callback
+            if cb is not None:
+                cb(self, ctx.getResponses())
+
+    def _processIncomingResponse(self, resp):
+        if isinstance(resp, list):
+            map(self._processIncomingResponse, resp)
+            return
+
+        resp = JsonRpcResponse.fromRawObject(resp)
+        with self._lock:
+            ctx = self._runningRequests.pop(resp.id)
+            ctx.addResponse(resp)
+
+        self._finalizeCtx(ctx)
+
+    def close(self):
+        self._transport.close()
+
+
 class JsonRpcServer(object):
     log = logging.getLogger("jsonrpc.JsonRpcServer")
 
diff --git a/yajsonrpc/asyncoreReactor.py b/yajsonrpc/asyncoreReactor.py
index 1ef6264..8c7fa5a 100644
--- a/yajsonrpc/asyncoreReactor.py
+++ b/yajsonrpc/asyncoreReactor.py
@@ -107,6 +107,15 @@
     def close(self):
         self._client.close()
 
+    def __eq__(self, other):
+        if type(self) != type(other):
+            return False
+
+        return hash(self) == hash(other)
+
+    def __hash__(self):
+        return hash(type(self)) ^ hash(self._client)
+
 
 # FIXME: We should go about making a listener wrapper like the client wrapper
 #        This is not as high priority as users don't interact with listeners
diff --git a/yajsonrpc/client.py b/yajsonrpc/client.py
deleted file mode 100644
index b7d2786..0000000
--- a/yajsonrpc/client.py
+++ /dev/null
@@ -1,45 +0,0 @@
-import json
-from yajsonrpc import \
-    JsonRpcError, \
-    asyncoreReactor
-
-_Size = asyncoreReactor._Size
-
-proton = None
-try:
-    import proton
-    from yajsonrpc import protonReactor
-    proton         # Squash pyflakes error for
-    protonReactor  # unused import
-except ImportError:
-    pass
-
-
-class JsonRpcClient(object):
-    def __init__(self, reactorClient):
-        self._transport = reactorClient
-
-    def connect(self):
-        self._transport.connect()
-
-    def callMethod(self, methodName, params=(), reqId=None, timeout=None):
-        msg = {'jsonrpc': '2.0',
-               'method': methodName,
-               'params': params,
-               'id': reqId}
-
-        self._transport.send(json.dumps(msg, 'utf-8'), timeout=timeout)
-        # Notifications have no repsonse
-        if reqId is None:
-            return
-
-        resp = self._transport.recv(timeout=timeout)
-        resp = json.loads(resp)
-        if resp.get('error') is not None:
-            raise JsonRpcError(resp['error']['code'],
-                               resp['error']['message'])
-
-        return resp.get('result')
-
-    def close(self):
-        self._transport.close()
diff --git a/yajsonrpc/protonReactor.py b/yajsonrpc/protonReactor.py
index 2796e24..5cf845f 100644
--- a/yajsonrpc/protonReactor.py
+++ b/yajsonrpc/protonReactor.py
@@ -27,6 +27,8 @@
 SERVER_AUTH = 2
 CLIENT_AUTH = 3
 
+MBUFF_SIZE = 10
+
 
 # Used for reactor coroutines
 class Return(object):
@@ -73,7 +75,9 @@
     def _openClientSession(self):
         host, port = self._address
         amqpAddress = "ampq://%s:%d/vdsm" % (host, port)
-        senderName = "Client (%s)" % (amqpAddress,)
+        senderName = "jsonrpc.ProtonClient %s (%s)" % (str(uuid.uuid4()),
+                                                       amqpAddress,)
+        self.log = logging.getLogger(senderName)
 
         self.connector = proton.pn_connector(self._reactor._driver,
                                              host, str(port), None)
@@ -116,12 +120,17 @@
     def _pushIncomingMessage(self, msg):
         try:
             self._inbox.put_nowait((self, msg))
+            self.log.debug("Message queued on inbox (%d)", id(self._inbox))
         except AttributeError:
             # Inbox not set
+            self.log.warn("Message missed since inbox was not set for "
+                          "this client")
             pass
 
     def _popPendingMessage(self):
-        return self._outbox.get_nowait()
+        msg = self._outbox.get_nowait()
+        self.log.debug("Pulling message from outbox")
+        return msg
 
     def setInbox(self, queue):
         self._inbox = queue
@@ -140,6 +149,7 @@
     log = logging.getLogger("jsonrpc.ProtonReactor")
 
     def __init__(self, deliveryTimeout=5):
+        self.log = logging.getLogger("jsonrpc.ProtonReactor (%d)" % id(self))
         self._isRunning = False
         self._coroutines = []
 
@@ -148,6 +158,7 @@
         self._deliveryTimeout = deliveryTimeout
         self._commandQueue = Queue()
         self._listeners = {}
+        self._wakeEv = Event()
 
     def _activate(self, connector, cond):
         self._scheduleOp(False, proton.pn_connector_activate,
@@ -166,7 +177,6 @@
         return int(timeout * 1000)
 
     def _waitDriverEvent(self, timeout=None):
-        self.log.debug("Waiting for events")
         timeout = self._convertTimeout(timeout)
         proton.pn_driver_wait(self._driver, timeout)
 
@@ -209,10 +219,8 @@
             self.log.debug("Authentication-PENDING")
 
     def _processConnectors(self):
-        connector = proton.pn_driver_connector(self._driver)
+        connector = proton.pn_connector_head(self._driver)
         while connector:
-            self.log.debug("Process Connector")
-
             # releaes any connector that has been closed
             if proton.pn_connector_closed(connector):
                 self.log.debug("Closing connector")
@@ -233,7 +241,7 @@
 
                 proton.pn_connector_process(connector)
 
-            connector = proton.pn_driver_connector(self._driver)
+            connector = proton.pn_connector_next(connector)
 
     def _initConnection(self, conn):
         if proton.pn_connection_state(conn) & proton.PN_LOCAL_UNINIT:
@@ -241,8 +249,7 @@
             proton.pn_connection_open(conn)
 
     def createClient(self, address):
-        client = ProtonClient(self, None, None, None, address)
-        return client
+        return ProtonClient(self, None, None, None, address)
 
     def _openPendingSessions(self, conn, connector):
         ssn = proton.pn_session_head(conn, proton.PN_LOCAL_UNINIT)
@@ -260,7 +267,6 @@
     def _openLinks(self, conn):
         link = proton.pn_link_head(conn, proton.PN_LOCAL_UNINIT)
         while link:
-            self.log.debug("Opening Link")
             proton.pn_terminus_copy(proton.pn_link_source(link),
                                     proton.pn_link_remote_source(link))
             proton.pn_terminus_copy(proton.pn_link_target(link),
@@ -278,17 +284,14 @@
 
             elif proton.pn_link_is_receiver(link):
                 self.log.debug("Opening Link to recv messages")
-                proton.pn_link_flow(link, 1)
                 proton.pn_link_open(link)
+                proton.pn_link_flow(link, MBUFF_SIZE)
 
             link = proton.pn_link_next(link, proton.PN_LOCAL_UNINIT)
 
     def _processDeliveries(self, conn, connector):
         delivery = proton.pn_work_head(conn)
         while delivery:
-            self.log.debug("Process delivery %s" %
-                           proton.pn_delivery_tag(delivery))
-
             if proton.pn_delivery_readable(delivery):
                 self._processIncoming(delivery, connector)
             elif proton.pn_delivery_writable(delivery):
@@ -297,32 +300,46 @@
             delivery = proton.pn_work_next(delivery)
 
     def _cleanDeliveries(self, conn):
-        link = proton.pn_link_head(conn, (proton.PN_LOCAL_ACTIVE))
-        while link:
+        def link_iter(conn):
+            link = proton.pn_link_head(conn, (proton.PN_LOCAL_ACTIVE))
+            while link:
+                yield link
+                link = proton.pn_link_next(link, (proton.PN_LOCAL_ACTIVE))
+
+        def delivery_iter(link):
             d = proton.pn_unsettled_head(link)
             while d:
-                _next = proton.pn_unsettled_next(d)
+                yield d
+                d = proton.pn_unsettled_next(d)
+
+        for link in link_iter(conn):
+            for d in delivery_iter(link):
+                ctx = proton.pn_delivery_get_context(d)
+                if isinstance(ctx, str):
+                    continue
+
                 disp = proton.pn_delivery_remote_state(d)
-                age = time.time() - proton.pn_delivery_get_context(d)
-                self.log.debug("Checking delivery")
+                age = time.time() - ctx
+                self.log.debug("Checking delivery (%s)",
+                               proton.pn_delivery_tag(d))
+
                 if disp and disp != proton.PN_ACCEPTED:
                     self.log.warn("Message was not accepted by remote end")
 
                 if disp and proton.pn_delivery_settled(d):
                     self.log.debug("Message settled by remote end")
                     proton.pn_delivery_settle(d)
+                    proton.pn_delivery_clear(d)
 
                 elif age > self._deliveryTimeout:
                     self.log.warn("Delivary not settled by remote host")
                     proton.pn_delivery_settle(d)
+                    proton.pn_delivery_clear(d)
 
                 elif proton.pn_link_state(link) & proton.PN_REMOTE_CLOSED:
                     self.log.warn("Link closed before settling message")
                     proton.pn_delivery_settle(d)
-
-                d = _next
-
-            link = proton.pn_link_next(link, (proton.PN_LOCAL_ACTIVE))
+                    proton.pn_delivery_clear(d)
 
     def _cleanLinks(self, conn):
         link = proton.pn_link_head(conn, (proton.PN_LOCAL_ACTIVE |
@@ -370,26 +387,22 @@
                 ctx.sender = sender
                 continue
 
-            if proton.pn_link_credit(sender) == 0:
-                self.log.debug("Not enough credit, waiting")
-                continue
+            while proton.pn_link_credit(sender) > 0:
+                try:
+                    data = ctx._popPendingMessage()
+                except Empty:
+                    break
+                else:
+                    msg = proton.Message()
+                    msg.body = data
+                    d = proton.pn_delivery(sender,
+                                           "delivery-%s" % str(uuid.uuid4()))
 
-            try:
-                data = ctx._popPendingMessage()
-            except Empty:
-                continue
-            else:
-                msg = proton.Message()
-                msg.body = data
-                self.log.debug("Creating delivery")
-                proton.pn_link_set_context(sender, msg.encode())
-
-                proton.pn_delivery(sender,
-                                   "response-delivery-%s" % str(uuid.uuid4()))
-        self.log.debug("Queued what I could")
+                    proton.pn_delivery_set_context(d, msg.encode())
+                    self.log.debug("Queueing delivery (%s)",
+                                   proton.pn_delivery_tag(d))
 
     def _serviceConnector(self, connector):
-        self.log.debug("Service Connector")
         conn = proton.pn_connector_connection(connector)
 
         self._initConnection(conn)
@@ -410,42 +423,37 @@
         link = proton.pn_delivery_link(delivery)
         ssn = proton.pn_link_session(link)
         msg = []
-        rc, buff = proton.pn_link_recv(link, 1024)
-        while rc >= 0:
-            msg.append(buff)
+        self.log.debug("Receiving '%s'", proton.pn_delivery_tag(delivery))
+        while True:
             rc, buff = proton.pn_link_recv(link, 1024)
+            msg.append(buff)
+            if rc == proton.PN_EOS:
+                break
 
         msg = ''.join(msg)
 
+        self.log.debug("Received '%s'", proton.pn_delivery_tag(delivery))
+        proton.pn_link_advance(link)
         proton.pn_delivery_update(delivery, proton.PN_ACCEPTED)
+        proton.pn_delivery_settle(delivery)
+
         msgObj = proton.Message()
         msgObj.decode(msg)
         ctx = proton.pn_session_get_context(ssn)
         ctx._pushIncomingMessage(msgObj.body)
 
-        proton.pn_delivery_settle(delivery)
-        proton.pn_link_advance(link)
-
         # if more credit is needed, grant it
         if proton.pn_link_credit(link) == 0:
-            proton.pn_link_flow(link, 1)
+            proton.pn_link_flow(link, MBUFF_SIZE)
 
     def _processOutgoing(self, delivery):
         link = proton.pn_delivery_link(delivery)
-        msg = proton.pn_link_get_context(link)
-        sent = proton.pn_link_send(link, msg)
-        if sent < 0:
-            self.log.warn("Problem sending message")
-        else:
-            msg = msg[sent:]
-            if len(msg) != 0:
-                self.log.debug("Delivery partial")
-                proton.pn_link_set_context(link, msg)
-            else:
-                self.log.debug("Delivery finished")
-                proton.pn_link_set_context(link, "")
-                proton.pn_delivery_set_context(delivery, time.time())
-                proton.pn_link_advance(link)
+        msg = proton.pn_delivery_get_context(delivery)
+        proton.pn_link_send(link, msg)
+        if proton.pn_link_advance(link):
+            self.log.debug("Delivery finished (%s)",
+                           proton.pn_delivery_tag(delivery))
+            proton.pn_delivery_set_context(delivery, time.time())
 
     def createListener(self, address, acceptHandler):
         host, port = address


--
To view, visit http://gerrit.ovirt.org/12304
To unsubscribe, visit http://gerrit.ovirt.org/settings

Gerrit-MessageType: newchange
Gerrit-Change-Id: Ie514c9029416a5869802f065b49f2da033ff3da6
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