Change in vdsm[master]: jsonrpc: Stomp support

smizrahi at redhat.com smizrahi at redhat.com
Mon Apr 14 09:57:31 UTC 2014


Saggi Mizrahi has uploaded a new change for review.

Change subject: jsonrpc: Stomp support
......................................................................

jsonrpc: Stomp support

The mini broker turned up to be much more work than I expected.
This is an intermediate solution. VDSM will accept broker like
message but will ignore most of them and not enforce policy.

This is so that we can get the engine talking in the correct way and
work on the mini broker stress free. This also means that we can release
VDSM without the mini broker using a message format that is future
ready.

It is expected from the engine to send the CONNECT and SUBSCRIBE frames
even though VDSM doesn't yet enforce them because future VDSM's probably
will.

Change-Id: I22bcae1e150dea7bc7d9fecefb6847c48bfe8949
Signed-off-by: Saggi Mizrahi <smizrahi at redhat.com>
---
M lib/yajsonrpc/__init__.py
M lib/yajsonrpc/betterAsyncore.py
A lib/yajsonrpc/stomp.py
A lib/yajsonrpc/stompReactor.py
M tests/jsonRpcTests.py
M tests/jsonRpcUtils.py
M vdsm_api/BindingJsonRpc.py
7 files changed, 895 insertions(+), 1 deletion(-)


  git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/50/26750/1

diff --git a/lib/yajsonrpc/__init__.py b/lib/yajsonrpc/__init__.py
index fabc5d4..34ff97f 100644
--- a/lib/yajsonrpc/__init__.py
+++ b/lib/yajsonrpc/__init__.py
@@ -466,6 +466,7 @@
         self._threadFactory = threadFactory
 
     def queueRequest(self, req):
+        self.log.debug("Queueing request")
         self._workQueue.put_nowait(req)
 
     def _serveRequest(self, ctx, req):
@@ -502,11 +503,14 @@
 
     def serve_requests(self):
         while True:
+            self.log.debug("Waiting for request")
             obj = self._workQueue.get()
+            self.log.debug("Popped request")
             if obj is None:
                 break
 
             client, msg = obj
+            self.log.debug("Parsing message")
             self._parseMessage(client, msg)
 
     def _parseMessage(self, client, msg):
@@ -560,4 +564,5 @@
             self._threadFactory(partial(self._serveRequest, ctx, request))
 
     def stop(self):
+        self.log.info("Stopping JsonRPC Server")
         self._workQueue.put_nowait(None)
diff --git a/lib/yajsonrpc/betterAsyncore.py b/lib/yajsonrpc/betterAsyncore.py
index 8806ee5..1e2c5ba 100644
--- a/lib/yajsonrpc/betterAsyncore.py
+++ b/lib/yajsonrpc/betterAsyncore.py
@@ -41,6 +41,9 @@
         self.connection = connection
         self.__wctx = wrappedContext
 
+    def pending(self):
+        return self.connection.pending()
+
     def get_context(self):
         return self.__wctx
 
diff --git a/lib/yajsonrpc/stomp.py b/lib/yajsonrpc/stomp.py
new file mode 100644
index 0000000..3e2ecfb
--- /dev/null
+++ b/lib/yajsonrpc/stomp.py
@@ -0,0 +1,522 @@
+import logging
+import socket
+import cStringIO
+from threading import Timer, Event
+from uuid import uuid4
+from collections import deque
+import time
+
+from betterAsyncore import Dispatcher
+import asyncore
+
+
+_ESCAPE_CHARS = (('\\\\', '\\'), ('\\r', '\r'), ('\\n', '\n'), ('\\c', ':'))
+
+
+class Command:
+    MESSAGE = "MESSAGE"
+    SEND = "SEND"
+    SUBSCRIBE = "SUBSCRIBE"
+    UNSUBSCRIBE = "UNSUBSCRIBE"
+    CONNECT = "CONNECT"
+    CONNECTED = "CONNECTED"
+    ERROR = "ERROR"
+    RECEIPT = "RECEIPT"
+
+
+class AckMode:
+    AUTO = "auto"
+
+
+class StompError(RuntimeError):
+    def __init__(self, frame):
+        RuntimeError.__init__(self, frame.body)
+
+
+class Frame(object):
+    def __init__(self, command="", headers=None, body=None):
+        self.command = command
+        if headers is None:
+            headers = {}
+
+        self.headers = headers
+        if isinstance(body, unicode):
+            body = body.encode('utf-8')
+
+        self.body = body
+
+    def encode(self):
+        body = self.body
+        # We do it here so we are sure header is up to date
+        if body is not None:
+            self.headers["content-length"] = len(body)
+
+        data = self.command + '\n'
+        data += '\n'.join(["%s:%s" % (encodeValue(key), encodeValue(value))
+                          for key, value in self.headers.iteritems()])
+        data += '\n\n'
+        if body is not None:
+            data += body
+
+        data += "\0"
+        return data
+
+    def __repr__(self):
+        return "<StompFrame command=%s>" % (repr(self.command))
+
+    def copy(self):
+        return Frame(self.command, self.headers.copy(), self.body)
+
+
+def decodeValue(s):
+    s = s.decode('utf-8')
+    for escaped, raw in _ESCAPE_CHARS:
+        s = s.replace(escaped, raw)
+
+    # TODO : Throw error on invalid escape char
+    # spec: Undefined escape sequences such as \t (octet 92 and 116) MUST be
+    # treated as a fatal protocol error.
+    return s
+
+
+def encodeValue(s):
+    if not isinstance(s, (str, unicode)):
+        s = str(s)
+    for escaped, raw in _ESCAPE_CHARS:
+        s = s.replace(raw, escaped)
+
+    if isinstance(s, unicode):
+        s = s.encode('utf-8')
+
+    return s
+
+
+class Parser(object):
+    _STATE_CMD = "Parsing command"
+    _STATE_HEADER = "Parsing headers"
+    _STATE_BODY = "Receiving body"
+
+    def __init__(self):
+        self._states = {
+            self._STATE_CMD: self._parse_command,
+            self._STATE_HEADER: self._parse_header,
+            self._STATE_BODY: self._parse_body}
+        self._frames = deque()
+        self._state = self._STATE_CMD
+        self._contentLength = -1
+        self._flush()
+
+    def _flush(self):
+        self._buffer = cStringIO.StringIO()
+
+    def _handle_terminator(self, term):
+        if term not in self._buffer.getvalue():
+            return None
+
+        data = self._buffer.getvalue()
+        res, rest = data.split(term, 1)
+        self._flush()
+        self._buffer.write(rest)
+
+        return res
+
+    def _parse_command(self):
+        cmd = self._handle_terminator('\n')
+        if cmd is None:
+            return False
+
+        if len(cmd) > 0 and cmd[-1] == '\r':
+            cmd = cmd[:-1]
+
+        if cmd == "":
+            return True
+
+        cmd = decodeValue(cmd)
+        self._tmpFrame = Frame(cmd)
+
+        self._state = self._STATE_HEADER
+        return True
+
+    def _parse_header(self):
+        header = self._handle_terminator('\n')
+        if header is None:
+            return False
+
+        headers = self._tmpFrame.headers
+        if len(header) > 0 and header[-1] == '\r':
+            header = header[:-1]
+
+        if header == "":
+            self._contentLength = int(headers.get('content-length', -1))
+            self._state = self._STATE_BODY
+            return True
+
+        # TODO: Proper error on missing or to many ':'
+        key, value = header.split(":")
+        key = decodeValue(key)
+        value = decodeValue(value)
+
+        # If a client or a server receives repeated frame header entries, only
+        # the first header entry SHOULD be used as the value of header entry.
+        # Subsequent values are only used to maintain a history of state
+        # changes of the header and MAY be ignored.
+        if key not in headers:
+            headers[key] = value
+
+        return True
+
+    def _pushFrame(self):
+        self._frames.append(self._tmpFrame)
+        self._state = self._STATE_CMD
+        self._tmpFrame = None
+        self._contentLength = -1
+
+    def _parse_body(self):
+        if self._contentLength >= 0:
+            return self._parse_body_length()
+        else:
+            return self._parse_body_terminator()
+
+    def _parse_body_terminator(self):
+        body = self._handle_terminator('\0')
+        if body is None:
+            return False
+
+        self._tmpFrame.body = body
+        self._pushFrame()
+        return True
+
+    def _parse_body_length(self):
+        buf = self._buffer
+        cl = self._contentLength
+        ndata = buf.tell()
+        if ndata < cl:
+            return False
+
+        remainingBytes = 0
+        self._flush()
+        body = buf.getvalue()
+        self._buffer.write(body[cl + 1:])
+        body = body[:cl]
+
+        if remainingBytes == 0:
+            self._tmpFrame.body = body
+            self._pushFrame()
+
+        return True
+
+    @property
+    def pending(self):
+        return len(self._frames)
+
+    def parse(self, data):
+        states = self._states
+        self._buffer.write(data)
+        while states[self._state]():
+            pass
+
+    def popFrame(self):
+        try:
+            return self._frames.popleft()
+        except IndexError:
+            return None
+
+
+class Client(object):
+    def __init__(self, sock=None):
+        """
+        Initialize the client.
+
+        The socket parameter can be an already initialized socket. Should be
+        used to pass specialized socket objects like SSL sockets.
+        """
+        if sock is None:
+            sock = self._sock = socket.socket()
+        else:
+            sock = sock
+
+        self._map = {}
+        # Because we don't know how large the frames are
+        # we have to use non bolocking IO
+        sock.setblocking(False)
+
+        # We have our own timeout for operations we
+        # pretend to be synchronous (like connect).
+        self._timeout = None
+        self._connected = Event()
+        self._subscriptions = {}
+
+        self._aclient = None
+        self._adisp = None
+
+        self._inbox = deque()
+
+    @property
+    def outgoing(self):
+        return self._adisp.outgoing
+
+    def _registerSubscription(self, sub):
+        self._subscriptions[sub.id] = sub
+
+    def _unregisterSubscription(self, sub):
+        del self._subscriptions[sub.id]
+
+    @property
+    def connected(self):
+        return self._connected.isSet()
+
+    def handle_connect(self, aclient, frame):
+        self._connected.set()
+
+    def handle_message(self, aclient, frame):
+        self._inbox.append(frame)
+
+    def process(self, timeout=None):
+        if timeout is None:
+            timeout = self._timeout
+
+        asyncore.loop(use_poll=True, timeout=timeout, map=self._map, count=1)
+
+    def connect(self, address, hostname):
+        sock = self._sock
+
+        self._aclient = AsyncClient(self, hostname)
+        adisp = self._adisp = AsyncDispatcher(self._aclient)
+        disp = self._disp = Dispatcher(adisp, sock, self._map)
+        sock.setblocking(True)
+        disp.connect(address)
+        sock.setblocking(False)
+        self.recv()  # wait for CONNECTED msg
+
+        if not self._connected.isSet():
+            sock.close()
+            raise socket.error()
+
+    def recv(self):
+        timeout = self._timeout
+        s = time.time()
+        duration = 0
+        while timeout is None or duration < timeout:
+            try:
+                return self._inbox.popleft()
+            except IndexError:
+                td = timeout - duration if timeout is not None else None
+                self.process(td)
+                duration = time.time() - s
+
+        return None
+
+    def put_subscribe(self, destination, ack=None):
+        subid = self._aclient.subscribe(self._adisp, destination, ack)
+        sub = Subsciption(self, subid, ack)
+        self._registerSubscription(sub)
+        return sub
+
+    def put_send(self, destination, data="", headers=None):
+        self._aclient.send(self._adisp, destination, data, headers)
+
+    def put(self, frame):
+        self._adisp.send_raw(frame)
+
+    def send(self):
+        disp = self._disp
+        timeout = self._timeout
+        duration = 0
+        s = time.time()
+        while ((timeout is None or duration < timeout) and
+               (disp.writable() or not self._connected.isSet())):
+                td = timeout - duration if timeout is not None else None
+                self.process(td)
+                duration = time.time() - s
+
+    def gettimout(self):
+        return self._timeout
+
+    def settimeout(self, value):
+        self._timeout = value
+
+
+class AsyncDispatcher(object):
+    log = logging.getLogger("stomp.AsyncDispatcher")
+
+    def __init__(self, frameHandler, bufferSize=1024):
+        self._frameHandler = frameHandler
+        self._bufferSize = bufferSize
+        self._parser = Parser()
+        self._outbox = deque()
+        self._outbuf = None
+
+    def _queueFrame(self, frame):
+        self._outbox.append(frame)
+
+    @property
+    def outgoing(self):
+        n = len(self._outbox)
+        if self._outbuf != "":
+            n += 1
+
+        return n
+
+    def handle_connect(self, dispatcher):
+        self._frameHandler.handle_connect(self)
+
+    def handle_read(self, dispatcher):
+        pending = self._bufferSize
+        while pending:
+            try:
+                data = dispatcher.recv(pending)
+            except socket.error:
+                dispatcher.handle_error()
+                return
+
+            try:
+                pending = dispatcher.socket.pending()
+            except AttributeError:
+                pending = 0
+                pass
+
+            parser = self._parser
+
+            if data is not None:
+                parser.parse(data)
+
+        frameHandler = self._frameHandler
+        if hasattr(frameHandler, "handle_frame"):
+            while parser.pending > 0:
+                frameHandler.handle_frame(self, parser.popFrame())
+
+    def popFrame(self):
+        return self._parser.popFrame()
+
+    def handle_write(self, dispatcher):
+        if self._outbuf is None:
+            try:
+                frame = self._outbox.popleft()
+            except IndexError:
+                return
+
+            self._outbuf = frame.encode()
+
+        data = self._outbuf
+        numSent = dispatcher.send(data)
+        if numSent == len(data):
+            self._outbuf = None
+        else:
+            self._outbuf = data[numSent:]
+
+    def send_raw(self, frame):
+        self._queueFrame(frame)
+
+    def writable(self, dispatcher):
+        return len(self._outbox) > 0 or self._outbuf is not None
+
+    def readable(self, dispatcher):
+        return True
+
+
+class AsyncClient(object):
+    log = logging.getLogger("yajsonrpc.protocols.stomp.AsyncClient")
+
+    def __init__(self, frameHandler, hostname):
+        self._hostname = hostname
+        self._frameHandler = frameHandler
+        self._connected = False
+        self._error = None
+        self._commands = {
+            Command.CONNECTED: self._process_connected,
+            Command.MESSAGE: self._process_message,
+            Command.RECEIPT: self._process_receipt,
+            Command.ERROR: self._process_error}
+
+    @property
+    def connected(self):
+        return self._connected
+
+    def getLastError(self):
+        return self._error
+
+    def handle_connect(self, dispatcher):
+        self.log.debug("Sending CONNECT frame...")
+        hostname = self._hostname
+        frame = Frame(
+            Command.CONNECT,
+            {"accept-version": "1.2",
+             "host": hostname})
+
+        dispatcher.send_raw(frame)
+
+    def handle_frame(self, dispatcher, frame):
+        self._commands[frame.command](frame, dispatcher)
+
+    def _process_connected(self, frame, dispatcher):
+        self._connected = True
+        frameHandler = self._frameHandler
+        if hasattr(frameHandler, "handle_connect"):
+            frameHandler.handle_connect(self, frame)
+
+        self.log.debug("Stomp connection astablished")
+
+    def _process_message(self, frame, dispatcher):
+        frameHandler = self._frameHandler
+
+        if hasattr(frameHandler, "handle_message"):
+            frameHandler.handle_message(self, frame)
+
+    def _process_receipt(self, frame, dispatcher):
+        # TODO:
+        pass
+
+    def _process_error(self, frame, dispatcher):
+        raise StompError(frame)
+
+    def send(self, dispatcher, destination, data="", headers=None):
+        frame = Frame(
+            Command.SEND,
+            {"destination": destination},
+            data)
+
+        dispatcher.send_raw(frame)
+
+    def subscribe(self, dispatcher, destination, ack=None):
+        if ack is None:
+            ack = AckMode.AUTO
+
+        subscriptionID = str(uuid4())
+
+        frame = Frame(
+            Command.SUBSCRIBE,
+            {"destination": destination,
+             "ack": ack,
+             "id": subscriptionID})
+
+        dispatcher.send_raw(frame)
+
+        return subscriptionID
+
+
+class Subsciption(object):
+    def __init__(self, client, subid, ack):
+        self._ack = ack
+        self._subid = subid
+        self._client = client
+        self._valid = True
+
+    @property
+    def id(self):
+        return self._subid
+
+    def unsubscribe(self):
+        client = self._client
+        subid = self._subid
+
+        client._unregisterSubscription(self)
+
+        frame = Frame(Command.UNSUBSCRIBE,
+                      {"id": str(subid)})
+        client.put(frame)
+        self._valid = False
+
+    def __del__(self):
+        # Using a timer because unsubscribe action might involve taking locks.
+        if self._valid:
+            Timer(0, self.unsubscribe)
diff --git a/lib/yajsonrpc/stompReactor.py b/lib/yajsonrpc/stompReactor.py
new file mode 100644
index 0000000..7448d76
--- /dev/null
+++ b/lib/yajsonrpc/stompReactor.py
@@ -0,0 +1,335 @@
+# Copyright (C) 2014 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 struct
+import asyncore
+import socket
+import os
+import threading
+import logging
+
+import stomp
+
+from betterAsyncore import \
+    Dispatcher, \
+    SSLDispatcher, \
+    SSLContext
+
+
+_Size = struct.Struct("!Q")
+
+_STATE_LEN = "Waiting for message length"
+_STATE_MSG = "Waiting for message"
+
+
+_DEFAULT_RESPONSE_DESTINATIOM = "/queue/_local/vdsm/reponses"
+_DEFAULT_REQUEST_DESTINATION = "/queue/_local/vdsm/requests"
+
+
+def _SSLContextFactory(ctxdef):
+    """Creates an appropriate ssl context from the generic defenition defined
+    in __init__.py"""
+    return SSLContext(cert_file=ctxdef.cert_file,
+                      key_file=ctxdef.key_file,
+                      ca_cert=ctxdef.ca_cert,
+                      session_id=ctxdef.session_id,
+                      protocol=ctxdef.protocol)
+
+
+class StompAdapterImpl(object):
+    log = logging.getLogger("Broker.StompAdapter")
+
+    def __init__(self, reactor, messageHandler):
+        self._reactor = reactor
+        self._messageHandler = messageHandler
+        self._commands = {
+            stomp.Command.CONNECT: self._cmd_connect,
+            stomp.Command.SEND: self._cmd_send,
+            stomp.Command.SUBSCRIBE: self._cmd_subscribe,
+            stomp.Command.UNSUBSCRIBE: self._cmd_unsubscribe}
+
+    def _cmd_connect(self, dispatcher, frame):
+        self.log.info("Processing CONNECT request")
+        version = frame.headers.get("accept-version", None)
+        if version != "1.2":
+            res = stomp.Frame(stomp.Command.ERROR, None, "Version unsupported")
+
+        else:
+            res = stomp.Frame(stomp.Command.CONNECTED, {"version": "1.2"})
+
+        dispatcher.send_raw(res)
+        self.log.info("CONNECT response queued")
+        self._reactor.wakeup()
+
+    def _cmd_subscribe(self, dispatcher, frame):
+        self.log.debug("Subscribe command ignored")
+
+    def _cmd_unsubscribe(self, dispatcher, frame):
+        self.log.debug("Unsubscribe command ignored")
+
+    def _cmd_send(self, dispatcher, frame):
+        self.log.debug("Passing incoming message")
+        self._messageHandler(self, frame.body)
+
+    def handle_frame(self, dispatcher, frame):
+        self.log.debug("Handling message %s", frame)
+        try:
+            self._commands[frame.command](dispatcher, frame)
+        except KeyError:
+            self.log.warn("Unknown command %s", frame)
+            dispatcher.handle_error()
+
+
+class StompServer(object):
+    log = logging.getLogger("yajsonrpc.StompServer")
+
+    def __init__(self, sock, reactor, addr, sslctx=None):
+        self._addr = addr
+        self._reactor = reactor
+        self._messageHandler = None
+
+        adapter = StompAdapterImpl(reactor, self._handleMessage)
+        adisp = self._adisp = stomp.AsyncDispatcher(adapter)
+        if sslctx is None:
+            try:
+                sslctx = sock.get_context()
+            except AttributeError:
+                # if get_context() is missing it just means we recieved an
+                # socket that doesn't use SSL
+                pass
+        else:
+            sslctx = _SSLContextFactory(sslctx)
+
+        if sslctx is None:
+            dispatcher = Dispatcher(adisp, sock=sock, map=reactor._map)
+        else:
+            dispatcher = SSLDispatcher(adisp, sslctx, sock=sock,
+                                       map=reactor._map)
+
+        if sock is None:
+            address_family = socket.getaddrinfo(*addr)[0][0]
+            dispatcher.create_socket(address_family, socket.SOCK_STREAM)
+
+        self._dispatcher = dispatcher
+
+    def setTimeout(self, timeout):
+        self._dispatcher.socket.settimeout(timeout)
+
+    def connect(self):
+        self._dispatcher.connect(self._addr)
+
+    def _handleMessage(self, impl, data):
+        if self._messageHandler is not None:
+            self.log.debug("Processing incoming message")
+            self._messageHandler((self, data))
+
+    def setMessageHandler(self, msgHandler):
+        self._messageHandler = msgHandler
+
+    def send(self, message):
+        self.log.debug("Sending response")
+        res = stomp.Frame(stomp.Command.MESSAGE,
+                          {"destination": _DEFAULT_RESPONSE_DESTINATIOM,
+                           "content-type": "application/json"},
+                          message)
+        self._adisp.send_raw(res)
+        self._reactor.wakeup()
+
+    def close(self):
+        self._dispatcher.close()
+
+
+class StompClient(object):
+    log = logging.getLogger("jsonrpc.AsyncoreClient")
+
+    def __init__(self, sock, reactor, addr, sslctx=None):
+        self._addr = addr
+        self._reactor = reactor
+        self._messageHandler = None
+
+        self._aclient = stomp.AsyncClient(self, "vdsm")
+        adisp = self._adisp = stomp.AsyncDispatcher(self._aclient)
+        if sslctx is None:
+            try:
+                sslctx = sock.get_context()
+            except AttributeError:
+                # if get_context() is missing it just means we recieved an
+                # socket that doesn't use SSL
+                pass
+        else:
+            sslctx = _SSLContextFactory(sslctx)
+
+        if sslctx is None:
+            dispatcher = Dispatcher(adisp, sock=sock, map=reactor._map)
+        else:
+            dispatcher = SSLDispatcher(adisp, sslctx, sock=sock,
+                                       map=reactor._map)
+
+        if sock is None:
+            address_family = socket.getaddrinfo(*addr)[0][0]
+            dispatcher.create_socket(address_family, socket.SOCK_STREAM)
+
+        self._dispatcher = dispatcher
+
+    def setTimeout(self, timeout):
+        self._dispatcher.socket.settimeout(timeout)
+
+    def connect(self):
+        self._dispatcher.connect(self._addr)
+
+    def handle_message(self, impl, frame):
+        if self._messageHandler is not None:
+            self.log.debug("Queueing incoming message")
+            self._messageHandler((self, frame.body))
+
+    def setMessageHandler(self, msgHandler):
+        self._messageHandler = msgHandler
+
+    def send(self, message):
+        self._aclient.send(self._adisp, _DEFAULT_REQUEST_DESTINATION,
+                           message,
+                           {"content-type": "application/json"})
+        self._reactor.wakeup()
+        self.log.debug("Message queued for delivery")
+
+    def close(self):
+        self._dispatcher.close()
+
+
+def StompListener(reactor, address, acceptHandler, sslctx=None):
+    impl = StompListenerImpl(reactor, address, acceptHandler)
+    if sslctx is None:
+        return Dispatcher(impl, map=reactor._map)
+    else:
+        sslctx = _SSLContextFactory(sslctx)
+        return SSLDispatcher(impl, sslctx, map=reactor._map)
+
+
+# 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
+#        as much
+class StompListenerImpl(object):
+    log = logging.getLogger("jsonrpc.StompListener")
+
+    def __init__(self, reactor, address, acceptHandler):
+        self._reactor = reactor
+        self._address = address
+        self._acceptHandler = acceptHandler
+
+    def init(self, dispatcher):
+        address_family = socket.getaddrinfo(*self._address)[0][0]
+        dispatcher.create_socket(address_family, socket.SOCK_STREAM)
+
+        dispatcher.set_reuse_addr()
+        dispatcher.bind(self._address)
+        dispatcher.listen(5)
+
+    def handle_accept(self, dispatcher):
+        try:
+            pair = dispatcher.accept()
+        except Exception as e:
+            self.log.exception(e)
+            raise
+        if pair is None:
+            return
+
+        sock, addr = pair
+        self.log.debug("Accepting connection from client "
+                       "at tcp://%s:%s", addr[0], addr[1])
+
+        client = StompServer(sock, self._reactor, addr)
+        self._acceptHandler(self, client)
+
+    def writable(self, dispatcher):
+        return False
+
+
+class _AsyncoreEvent(asyncore.file_dispatcher):
+    def __init__(self, map):
+        self._lock = threading.Lock()
+        r, w = os.pipe()
+        self._w = w
+        try:
+            asyncore.file_dispatcher.__init__(self, r, map=map)
+        except:
+            os.close(r)
+            os.close(w)
+            raise
+
+        # the file_dispatcher ctor dups the file in order to take ownership of
+        # it
+        os.close(r)
+        self._isSet = False
+
+    def writable(self):
+        return False
+
+    def set(self):
+        with self._lock:
+            if self._isSet:
+                return
+
+            self._isSet = True
+
+        os.write(self._w, "a")
+
+    def handle_read(self):
+        with self._lock:
+            self.recv(1)
+            self._isSet = False
+
+    def close(self):
+        try:
+            os.close(self._w)
+        except (OSError, IOError):
+            pass
+
+        asyncore.file_dispatcher.close(self)
+
+
+class StompReactor(object):
+    def __init__(self, sslctx=None):
+        self.sslctx = sslctx
+        self._map = {}
+        self._isRunning = False
+        self._wakeupEvent = _AsyncoreEvent(self._map)
+
+    def createListener(self, address, acceptHandler):
+        listener = StompListener(self, address, acceptHandler, self.sslctx)
+        self.wakeup()
+        return listener
+
+    def createClient(self, address):
+        return StompClient(None, self, address, self.sslctx)
+
+    def process_requests(self):
+        self._isRunning = True
+        while self._isRunning:
+            asyncore.loop(use_poll=True, map=self._map, count=1)
+
+        for key, dispatcher in self._map.items():
+            del self._map[key]
+            dispatcher.close()
+
+    def wakeup(self):
+        self._wakeupEvent.set()
+
+    def stop(self):
+        self._isRunning = False
+        try:
+            self.wakeup()
+        except (IOError, OSError):
+            # Client woke up and closed the event dispatcher without our help
+            pass
diff --git a/tests/jsonRpcTests.py b/tests/jsonRpcTests.py
index 3b63c01..3eb210b 100644
--- a/tests/jsonRpcTests.py
+++ b/tests/jsonRpcTests.py
@@ -149,7 +149,10 @@
 
 
 class _DummyBridge(object):
+    log = logging.getLogger("tests.DummyBridge")
+
     def echo(self, text):
+        self.log.info("ECHO: '%s'", text)
         return text
 
     def ping(self):
@@ -185,6 +188,7 @@
         bridge = _DummyBridge()
         with constructServer(rt, bridge, ssl) as (server, clientFactory):
             with self._client(clientFactory) as client:
+                self.log.info("Calling 'echo'")
                 self.assertEquals(self._callTimeout(client, "echo", (data,),
                                   apiTests.id,
                                   CALL_TIMEOUT), data)
diff --git a/tests/jsonRpcUtils.py b/tests/jsonRpcUtils.py
index 6d6dcf2..e3f54f6 100644
--- a/tests/jsonRpcUtils.py
+++ b/tests/jsonRpcUtils.py
@@ -10,6 +10,7 @@
 from yajsonrpc import \
     JsonRpcServer, \
     asyncoreReactor, \
+    stompReactor, \
     JsonRpcClientPool, \
     SSLContext
 
@@ -36,6 +37,14 @@
 
 
 @contextmanager
+def _stompServerConstructor():
+    port = getFreePort()
+    address = ("127.0.0.1", port)
+    reactorType = stompReactor.StompReactor
+    yield reactorType, address
+
+
+ at contextmanager
 def _tcpServerConstructor():
     port = getFreePort()
     address = ("127.0.0.1", port)
@@ -56,7 +65,8 @@
 
 
 REACTOR_CONSTRUCTORS = {"tcp": _tcpServerConstructor,
-                        "amqp": _protonServerConstructor}
+                        "amqp": _protonServerConstructor,
+                        "stomp": _stompServerConstructor}
 REACTOR_TYPE_PERMUTATIONS = [[r] for r in REACTOR_CONSTRUCTORS.iterkeys()]
 CONNECTION_PERMUTATIONS = tuple(product(REACTOR_CONSTRUCTORS.iterkeys(),
                                         (True, False)))
diff --git a/vdsm_api/BindingJsonRpc.py b/vdsm_api/BindingJsonRpc.py
index 6c0fa47..471cb70 100644
--- a/vdsm_api/BindingJsonRpc.py
+++ b/vdsm_api/BindingJsonRpc.py
@@ -21,6 +21,7 @@
 
 from yajsonrpc import JsonRpcServer
 from yajsonrpc.asyncoreReactor import AsyncoreReactor
+from yajsonrpc.stompReactor import StompReactor
 from yajsonrpc.betterAsyncore import SSLContext
 ProtonReactor = None
 try:
@@ -49,6 +50,9 @@
             if backendType not in reactors:
                 if backendType == "tcp":
                     reactors["tcp"] = self._createTcpReactor(truststore_path)
+                elif backendType == "stomp":
+                    reactors["stomp"] = \
+                        self._createStompReactor(truststore_path)
                 elif backendType == "amqp":
                     if ProtonReactor is None:
                         continue
@@ -84,6 +88,15 @@
             ca_cert = truststore_path + '/certs/cacert.pem'
             return AsyncoreReactor(SSLContext(cert_file, key_file, ca_cert))
 
+    def _createStompReactor(self, truststore_path=None):
+        if truststore_path is None:
+            return StompReactor()
+        else:
+            key_file = truststore_path + '/keys/vdsmkey.pem'
+            cert_file = truststore_path + '/certs/vdsmcert.pem'
+            ca_cert = truststore_path + '/certs/cacert.pem'
+            return StompReactor(SSLContext(cert_file, key_file, ca_cert))
+
     def _createProtonReactor(self):
         return ProtonReactor()
 
@@ -104,6 +117,8 @@
             try:
                 if backendType == "tcp":
                     self._createTcpListener(cfg)
+                if backendType == "stomp":
+                    self._createStompListener(cfg)
                 elif backendType == "amqp":
                     self._createProtonListener(cfg)
             except:


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

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