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