Change in vdsm[master]: [WIP] AMQP support

smizrahi at redhat.com smizrahi at redhat.com
Tue Dec 11 21:38:33 UTC 2012


Saggi Mizrahi has uploaded a new change for review.

Change subject: [WIP] AMQP support
......................................................................

[WIP] AMQP support

Change-Id: I850005e7375472bbf6238fd38b10d8f1b5e2a191
Signed-off-by: Saggi Mizrahi <smizrahi at redhat.com>
---
M tests/jsonRpcTests.py
A vdsm_api/jsonrpc/protonReactor.py
2 files changed, 380 insertions(+), 1 deletion(-)


  git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/87/9987/1

diff --git a/tests/jsonRpcTests.py b/tests/jsonRpcTests.py
index 85b98a6..dd9148f 100644
--- a/tests/jsonRpcTests.py
+++ b/tests/jsonRpcTests.py
@@ -22,12 +22,22 @@
 from contextlib import contextmanager
 from functools import partial
 from contextlib import closing
+import uuid
+
+from nose.plugins.skip import SkipTest
 
 from testrunner import VdsmTestCase as TestCaseBase, \
     expandPermutations, \
     permutations
 
 from jsonrpc import tcpReactor
+
+protonReactor = None
+try:
+    import proton
+    from jsonrpc import protonReactor
+except ImportError:
+    pass
 
 PORT_RANGE = xrange(49152, 65535)
 
@@ -65,7 +75,22 @@
         reactor.stop()
 
 
-REACTOR_CONSTRUCTORS = {"tcp": _tcpServerConstructor}
+ at contextmanager
+def _protonServerConstructor(messageHandler):
+    if protonReactor is None:
+        raise SkipTest("qpid-proton python bindings are not installed")
+
+    port = _getFreePort()
+    serverAddress = "amqp://127.0.0.1:%d" % (port,)
+    reactor = protonReactor.ProtonReactor(("127.0.0.1", port), messageHandler)
+
+    try:
+        yield reactor, partial(ProtonReactorClient, serverAddress)
+    finally:
+        reactor.stop()
+
+REACTOR_CONSTRUCTORS = {"tcp": _tcpServerConstructor,
+                        "proton": _protonServerConstructor}
 REACTOR_TYPE_PERMUTATIONS = [[r] for r in REACTOR_CONSTRUCTORS.iterkeys()]
 
 
@@ -103,6 +128,58 @@
         self.sock.close()
 
 
+class ProtonReactorClient(object):
+    def __init__(self, serverAddress):
+        self._serverAddress = serverAddress
+        self._msngr = proton.Messenger("client-%s" % str(uuid.uuid4()))
+
+    def connect(self):
+        self._msngr.start()
+
+    def sendMessage(self, data, timeout=None):
+        if timeout is None:
+            timeout = -1
+        else:
+            timeout *= 1000
+
+        msg = proton.Message()
+        msg.address = self._serverAddress
+        msg.body = unicode(data)
+        self._msngr.timeout = timeout
+        t = self._msngr.put(msg)
+        try:
+            self._msngr.send()
+        except:
+            self._msngr.settle(t)
+            raise
+
+    def recvMessage(self, timeout=None):
+        if timeout is None:
+            timeout = -1
+        else:
+            timeout *= 1000
+
+        self._msngr.timeout = timeout
+        self._msngr.recv(10)
+
+        if not self._msngr.incoming:
+            raise socket.timeout()
+
+        if self._msngr.incoming > 1:
+            raise Exception("Got %d repsones instead of 1" %
+                            self._msngr.incoming)
+
+        msg = proton.Message()
+        self._msngr.get(msg)
+        return msg.body
+
+    def close(self):
+        try:
+            self._msngr.stop()
+        except:
+            pass
+
+
 @expandPermutations
 class ReactorTests(TestCaseBase):
     @permutations(REACTOR_TYPE_PERMUTATIONS)
diff --git a/vdsm_api/jsonrpc/protonReactor.py b/vdsm_api/jsonrpc/protonReactor.py
new file mode 100644
index 0000000..2138427
--- /dev/null
+++ b/vdsm_api/jsonrpc/protonReactor.py
@@ -0,0 +1,302 @@
+# Copyright (C) 2012 Saggi Mizrahi, Red Hat Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# General Public License for more details.
+#
+# You should have received a copy of the GNU General Public
+# License along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
+
+import threading
+import logging
+import uuid
+from Queue import Queue, Empty
+
+import proton
+
+FAILED = 0
+CONNECTED = 1
+AUTHENTICATING = 2
+
+
+class ProtonContext(object):
+    log = logging.getLogger("jsonrpc.ProtonContext")
+
+    def __init__(self, reactor, msg):
+        self._reactor = reactor
+        self._msg = msg
+
+    @property
+    def data(self):
+        return self._msg.body
+
+    def sendReply(self, data):
+        self._reactor.sendReply(self, data)
+
+
+class ProtonReactor(object):
+    log = logging.getLogger("jsonrpc.ProtonReactor")
+
+    def __init__(self, address, messageHandler):
+        self._messageHandler = messageHandler
+        host, port = address
+        self.host = host
+        self.port = port
+
+        self._isRunning = False
+
+        self._hasData = threading.Event()
+
+        self.driver = proton.pn_driver()
+        self.listener = proton.pn_listener(self.driver, self.host,
+                                           str(self.port), None)
+        if self.listener is None:
+            raise RuntimeError("Could not listen on %s:%s" % (self.host,
+                                                              self.port))
+
+        self._outQueue = Queue()
+        self._outMesseges = []
+
+    def _convertTimeout(self, timeout):
+        """
+        Timeouts in puthon are usually floats representing seconds, this
+        converts the conventional python timeout to proton compatible
+        millisecond timeouts
+        """
+
+        if timeout is None:
+            return -1
+
+        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)
+
+    def _acceptConnectionRequests(self):
+        l = proton.pn_driver_listener(self.driver)
+        while l:
+            self.log.debug("Accepting Connection.")
+            cxtr = proton.pn_listener_accept(l)
+            proton.pn_connector_set_context(cxtr, AUTHENTICATING)
+
+            l = proton.pn_driver_listener(self.driver)
+
+    def _authenticateConnector(self, cxtr):
+        self.log.debug("Authenticating...")
+        sasl = proton.pn_connector_sasl(cxtr)
+        state = proton.pn_sasl_state(sasl)
+        while state == proton.PN_SASL_CONF or state == proton.PN_SASL_STEP:
+            if state == proton.PN_SASL_CONF:
+                self.log.debug("Authenticating-CONF...")
+                proton.pn_sasl_mechanisms(sasl, "ANONYMOUS")
+                proton.pn_sasl_server(sasl)
+            elif state == proton.PN_SASL_STEP:
+                self.log.debug("Authenticating-STEP...")
+                mech = proton.pn_sasl_remote_mechanisms(sasl)
+                if mech == "ANONYMOUS":
+                    proton.pn_sasl_done(sasl, proton.PN_SASL_OK)
+                else:
+                    proton.pn_sasl_done(sasl, proton.PN_SASL_AUTH)
+            state = proton.pn_sasl_state(sasl)
+
+        if state == proton.PN_SASL_PASS:
+            proton.pn_connector_set_connection(cxtr, proton.pn_connection())
+            proton.pn_connector_set_context(cxtr, CONNECTED)
+            self.log.debug("Authentication-PASSED")
+        elif state == proton.PN_SASL_FAIL:
+            proton.pn_connector_set_context(cxtr, FAILED)
+            self.log.debug("Authentication-FAILED")
+        else:
+            self.log.debug("Authentication-PENDING")
+
+    def _processConnectors(self):
+        cxtr = proton.pn_driver_connector(self.driver)
+        while cxtr:
+            self.log.debug("Process Connector")
+
+            # releaes any connector that has been closed
+            if proton.pn_connector_closed(cxtr):
+                self.log.debug("Closing connector")
+                proton.pn_connector_free(cxtr)
+            else:
+                proton.pn_connector_process(cxtr)
+
+                state = proton.pn_connector_context(cxtr)
+                if state == AUTHENTICATING:
+                    self._authenticateConnector(cxtr)
+                elif state == CONNECTED:
+                    self._serviceConnector(cxtr)
+                else:
+                    self.log.warning("Unknown Connection state '%s'" % state)
+
+                proton.pn_connector_process(cxtr)
+
+            cxtr = proton.pn_driver_connector(self.driver)
+
+    def _initConnection(self, conn):
+        if proton.pn_connection_state(conn) & proton.PN_LOCAL_UNINIT:
+            self.log.debug("Connection Opened.")
+            proton.pn_connection_open(conn)
+
+    def _openPendingSessions(self, conn):
+        ssn = proton.pn_session_head(conn, proton.PN_LOCAL_UNINIT)
+        while ssn:
+            proton.pn_session_open(ssn)
+            self.log.debug("Session Opened.")
+            ssn = proton.pn_session_next(ssn, proton.PN_LOCAL_UNINIT)
+
+    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),
+                                    proton.pn_link_remote_target(link))
+
+            if proton.pn_link_is_sender(link):
+                linkSource = proton.pn_link_source(link)
+                addr = proton.pn_terminus_get_address(linkSource)
+
+                self.log.debug("Opening Link to send data to '%s'" % addr)
+                proton.pn_delivery(link,
+                                   "VDSM-delivery-%d" % str(uuid.uuid4()))
+            else:
+                linkTarget = proton.pn_link_target(link)
+                addr = proton.pn_terminus_get_address(linkTarget)
+                self.log.debug("Opening Link to recv messages from '%s'" %
+                               addr)
+                proton.pn_link_flow(link, 1)
+
+            proton.pn_link_open(link)
+            link = proton.pn_link_next(link, proton.PN_LOCAL_UNINIT)
+
+    def _distributeMessages(self):
+        while True:
+            try:
+                msg = self._outQueue.get_nowait()
+            except Empty:
+                break
+
+            self._outMesseges.append(msg)
+
+    def _processDeliveries(self, conn):
+        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)
+            elif proton.pn_delivery_writable(delivery):
+                self._processOutgoing(delivery)
+
+            delivery = proton.pn_work_next(delivery)
+
+    def _cleanLinks(self, conn):
+        link = proton.pn_link_head(conn, (proton.PN_LOCAL_ACTIVE |
+                                          proton.PN_REMOTE_CLOSED))
+        while link:
+            proton.pn_link_close(link)
+            link = proton.pn_link_next(link, (proton.PN_LOCAL_ACTIVE |
+                                              proton.PN_REMOTE_CLOSED))
+
+    def _cleanSessions(self, conn):
+        ssn = proton.pn_session_head(conn, (proton.PN_LOCAL_ACTIVE |
+                                            proton.PN_REMOTE_CLOSED))
+        while ssn:
+            proton.pn_session_close(ssn)
+            self.log.debug("Session Closed")
+            ssn = proton.pn_session_next(ssn, (proton.PN_LOCAL_ACTIVE |
+                                               proton.PN_REMOTE_CLOSED))
+
+    def _teardownConnection(self, conn):
+        if proton.pn_connection_state(conn) == ((proton.PN_LOCAL_ACTIVE |
+                                                 proton.PN_REMOTE_CLOSED)):
+            proton.pn_connection_close(conn)
+
+    def _serviceConnector(self, cxtr):
+        self.log.debug("Service Connector")
+        conn = proton.pn_connector_connection(cxtr)
+        self._initConnection(conn)
+        self._openPendingSessions(conn)
+        self._openLinks(conn)
+        self._processDeliveries(conn)
+        self._cleanLinks(conn)
+
+        conn = proton.pn_connector_connection(cxtr)
+
+    def _processIncoming(self, delivery):
+        link = proton.pn_delivery_link(delivery)
+        msg = []
+        rc, buff = proton.pn_link_recv(link, 1024)
+        while rc >= 0:
+            msg.append(buff)
+            rc, buff = proton.pn_link_recv(link, 1024)
+
+        msg = ''.join(msg)
+
+        proton.pn_delivery_update(delivery, proton.PN_ACCEPTED)
+        msgObj = proton.Message()
+        msgObj.decode(msg)
+        self._messageHandler.handleMessage(ProtonContext(self, msgObj))
+
+        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)
+
+    def _processOutgoing(self, delivery):
+        link = proton.pn_delivery_link(delivery)
+        mbox = proton.pn_terminus_get_address(proton.pn_link_source(link))
+        print mbox
+        self.log.debug("Request for Mailbox=%s" % str(mbox))
+        if mbox in self.mailboxes and self.mailboxes[mbox]:
+            msg = self.queuedMessages[mbox].pop(0)
+            self.log.debug("Fetched message %s" % str(msg))
+            msg = "%s=%s" % (mbox, msg)
+        else:
+            print("Warning: mailbox %s is empty, sending empty message." % mbox)
+            msg = "<EMPTY>"
+        sent = pn_link_send(link, msg)
+        assert(sent == len(msg))
+        self.log.debug("Msg Sent %d" % sent);
+
+        # if the link can accept more ???RAFI, is that correct????, grant
+        # another delivery
+        if pn_link_advance(link):
+            pn_delivery(link, "server-delivery-%d" % self.counter)
+            self.counter += 1
+
+        # do not settle the delivery now - wait until the remote sets the disposition.
+
+    def serve_forever(self):
+        self._isRunning = True
+        while self._isRunning:
+            self._waitDriverEvent()
+            self._acceptConnectionRequests()
+            self._processConnectors()
+
+    def sendReply(self, ctx, data):
+        msg = proton.Message()
+        msg.address = ctx._msg.reply_to
+        print msg.address
+        msg.body = data
+        self._outQueue.put_nowait(msg)
+
+    def stop(self):
+        #TODO: proper cleanup, release all connections
+        self._isRunning = False
+        #proton.pn_listener_close(self.listener)
+        #proton.pn_driver_free(self.driver)
+        self._isRunning = False


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

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