Change in vdsm[master]: subscriptions
piotr.kliczewski at gmail.com
piotr.kliczewski at gmail.com
Fri Mar 6 15:48:07 UTC 2015
Piotr Kliczewski has uploaded a new change for review.
Change subject: subscriptions
......................................................................
subscriptions
Change-Id: I1493070f2ba66ca9d39a6661876c82c4727cad62
Signed-off-by: pkliczewski <piotr.kliczewski at gmail.com>
---
M lib/stompClient.py
M lib/yajsonrpc/__init__.py
M lib/yajsonrpc/stomp.py
M lib/yajsonrpc/stompReactor.py
M tests/jsonRpcHelper.py
M vdsm/rpc/BindingJsonRpc.py
6 files changed, 177 insertions(+), 80 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/51/38451/1
diff --git a/lib/stompClient.py b/lib/stompClient.py
index 5a6d7f5..db6f3d0 100644
--- a/lib/stompClient.py
+++ b/lib/stompClient.py
@@ -10,6 +10,7 @@
import socket
from threading import Thread, Lock
import time
+import uuid
_reactor = None
_reactorLock = Lock()
@@ -61,7 +62,7 @@
stomp_client = reactor.createClient(sock)
subscription = stomp_client.subscribe(
_DEFAULT_RESPONSE_DESTINATION,
- sub_id="__vdsm_fake_broker__",
+ sub_id=str(uuid.uuid4()),
)
client = yjrpc.JsonRpcClient(
ClientRpcTransportAdapter(
diff --git a/lib/yajsonrpc/__init__.py b/lib/yajsonrpc/__init__.py
index 0bf3a54..f12d19b 100644
--- a/lib/yajsonrpc/__init__.py
+++ b/lib/yajsonrpc/__init__.py
@@ -205,9 +205,10 @@
class _JsonRpcServeRequestContext(object):
- def __init__(self, client):
+ def __init__(self, client, conn):
self._requests = []
self.client = client
+ self.conn = conn
self._counter = 0
self._requests = {}
self._responses = []
@@ -441,7 +442,7 @@
try:
params = req.params
- server_address = ctx.client.get_local_address()
+ server_address = ctx.conn.get_local_address()
self._bridge.register_server_address(server_address)
if isinstance(req.params, list):
res = method(*params)
@@ -469,11 +470,11 @@
if obj is None:
break
- client, msg = obj
- self._parseMessage(client, msg)
+ client, conn, msg = obj
+ self._parseMessage(client, conn, msg)
- def _parseMessage(self, client, msg):
- ctx = _JsonRpcServeRequestContext(client)
+ def _parseMessage(self, client, conn, msg):
+ ctx = _JsonRpcServeRequestContext(client, conn)
try:
rawRequests = json.loads(msg)
diff --git a/lib/yajsonrpc/stomp.py b/lib/yajsonrpc/stomp.py
index 0dd6617..8b0a296 100644
--- a/lib/yajsonrpc/stomp.py
+++ b/lib/yajsonrpc/stomp.py
@@ -21,8 +21,9 @@
from vdsm.utils import monotonic_time
import re
+LEGACY_SUB_ID_REQ = "/queue/_local/vdsm/requests"
+LEGACY_SUB_ID_RES = "/queue/_local/vdsm/reponses"
-FAKE_SUB_ID = "__vdsm_fake_broker__"
_RE_ESCAPE_SEQUENCE = re.compile(r"\\(.)")
@@ -296,8 +297,9 @@
class AsyncDispatcher(object):
log = logging.getLogger("stomp.AsyncDispatcher")
- def __init__(self, frame_handler, bufferSize=4096):
+ def __init__(self, connection, frame_handler, bufferSize=4096):
self._frame_handler = frame_handler
+ self.connection = connection
self._bufferSize = bufferSize
self._parser = Parser()
self._outbuf = None
@@ -445,11 +447,6 @@
def _process_message(self, frame, dispatcher):
sub_id = frame.headers.get(Headers.SUBSCRIPTION)
- if sub_id is None:
- # This is actually an error, but because of the fake broker we
- # injects a fake sub id
- sub_id = "__vdsm_fake_broker__"
-
sub = self._subscriptions.get(sub_id)
if sub is None:
self.log.warning(
@@ -458,7 +455,7 @@
)
return
- sub._handle_message(frame)
+ sub.handle_message(frame)
def _process_receipt(self, frame, dispatcher):
self.log.warning("Receipt frame received and ignored")
@@ -529,7 +526,7 @@
self._message_handler = message_handler
self._destination = destination
- def _handle_message(self, frame):
+ def handle_message(self, frame):
self._message_handler(self, frame)
def set_message_handler(self, handler):
diff --git a/lib/yajsonrpc/stompReactor.py b/lib/yajsonrpc/stompReactor.py
index 9ea45c0..c94fdc7 100644
--- a/lib/yajsonrpc/stompReactor.py
+++ b/lib/yajsonrpc/stompReactor.py
@@ -15,9 +15,11 @@
import logging
from collections import deque
+from uuid import uuid4
+import json
import stomp
-
+from vdsm import utils
from betterAsyncore import Dispatcher, Reactor
from vdsm.sslutils import SSLSocket
from . import (
@@ -55,10 +57,12 @@
class StompAdapterImpl(object):
log = logging.getLogger("Broker.StompAdapter")
- def __init__(self, reactor, messageHandler):
+ def __init__(self, reactor, sub_map, req_dest):
self._reactor = reactor
self._outbox = deque()
- self._messageHandler = messageHandler
+ self._sub_dests = sub_map
+ self._req_dest = req_dest
+ self._sub_ids = {}
self._commands = {
stomp.Command.CONNECT: self._cmd_connect,
stomp.Command.SEND: self._cmd_send,
@@ -106,13 +110,99 @@
self._reactor.wakeup()
def _cmd_subscribe(self, dispatcher, frame):
- self.log.debug("Subscribe command ignored")
+ self.log.debug("Subscribe command received")
+ destination = frame.headers.get("destination", None)
+ sub_id = frame.headers.get("id", None)
+
+ if not destination or not sub_id:
+ self._send_error("Missing required header",
+ dispatcher.connection)
+ return
+
+ ack = frame.headers.get("ack", stomp.AckMode.AUTO)
+ subscription = stomp.Subscription(dispatcher.connection, destination,
+ sub_id, ack, None)
+
+ # TODO make sure to cleanup when connection lost
+ try:
+ subs = self._sub_dests[destination]
+ except KeyError:
+ self._sub_dests[destination] = [subscription]
+ else:
+ subs.append(subscription)
+
+ self._sub_ids[sub_id] = subscription
+
+ def _send_error(self, msg, connection):
+ res = stomp.Frame(
+ stomp.Command.ERROR,
+ None,
+ msg
+ )
+ connection.send_raw(res)
def _cmd_unsubscribe(self, dispatcher, frame):
- self.log.debug("Unsubscribe command ignored")
+ self.log.debug("Unsubscribe command received")
+ sub_id = frame.headers.get("id", None)
+
+ if not sub_id:
+ self._send_error("Missing required header",
+ dispatcher.connection)
+ return
+
+ subscription = self.subs_ids[sub_id]
+ if not subscription:
+ # ignore
+ return
+
+ del self.subs_ids[sub_id]
+ subs = self._sub_dests[subscription.destination]
+ if len(subs) == 1:
+ del self._sub_dests[subscription.destination]
+ else:
+ if subscription in subs:
+ subs.remove(subscription)
def _cmd_send(self, dispatcher, frame):
- self._messageHandler(self, frame.body)
+ destination = frame.headers.get(stomp.Headers.DESTINATION, None)
+ if _DEFAULT_REQUEST_DESTINATION == destination:
+ # default subscription
+ try:
+ reqs = json.loads(frame.body)
+ self._req_dest[reqs[0].get("id")] = frame.headers.get(
+ stomp.Headers.REPLY_TO
+ )
+ except:
+ # let the jsonrpc server to reply
+ pass
+
+ dispatcher.connection.handleMessage(frame.body)
+ return
+ elif stomp.LEGACY_SUB_ID_REQ == destination:
+ try:
+ reqs = json.loads(frame.body)
+ self._req_dest[reqs[0].get("id")] = stomp.LEGACY_SUB_ID_RES
+ except:
+ # let the jsonrpc server to reply
+ pass
+ dispatcher.connection.handleMessage(frame.body)
+ return
+
+ subs = self._sub_dests[destination]
+ if len(subs) == 0:
+ self._send_error("Subscription not available",
+ dispatcher.connection)
+ return
+
+ for subscription in subs:
+ headers = utils.picklecopy(frame.headers)
+ headers[stomp.Headers.SUBSCRIPTION] = subscription.id
+ res = stomp.Frame(
+ stomp.Command.MESSAGE,
+ headers,
+ frame.body
+ )
+ subscription.client.send_raw(res)
def handle_frame(self, dispatcher, frame):
self.log.debug("Handling message %s", frame)
@@ -124,13 +214,15 @@
class _StompConnection(object):
- def __init__(self, aclient, sock, reactor):
+ log = logging.getLogger("yajsonrpc.StompServer")
+
+ def __init__(self, server, aclient, sock, reactor):
self._socket = sock
self._reactor = reactor
- self._messageHandler = None
+ self._server = server
self._aclient = aclient
- adisp = self._adisp = stomp.AsyncDispatcher(aclient)
+ adisp = self._adisp = stomp.AsyncDispatcher(self, aclient)
self._dispatcher = Dispatcher(adisp, sock=sock, map=reactor._map)
def send_raw(self, msg):
@@ -146,38 +238,32 @@
def close(self):
self._dispatcher.close()
+ def get_local_address(self):
+ return self._socket.getsockname()[0]
+
+ def setMessageHandler(self, msgHandler):
+ self._messageHandler = msgHandler
+ self._dispatcher.handle_read_event()
+
+ def handleMessage(self, data):
+ if self._messageHandler is not None:
+ self._messageHandler((self._server, self, data))
+
class StompServer(object):
log = logging.getLogger("yajsonrpc.StompServer")
- def __init__(self, sock, reactor):
+ def __init__(self, reactor):
self._reactor = reactor
self._messageHandler = None
- self._socket = sock
+ self._sub_map = {}
+ self._req_dest = {}
- adapter = StompAdapterImpl(reactor, self._handleMessage)
- self._stompConn = _StompConnection(
- adapter,
- sock,
- reactor,
- )
-
- def setTimeout(self, timeout):
- self._stompConn.setTimeout(timeout)
-
- def connect(self):
- self._stompConn.connect()
-
- def _handleMessage(self, impl, data):
- if self._messageHandler is not None:
- self._messageHandler((self, data))
-
- def setMessageHandler(self, msgHandler):
- self._messageHandler = msgHandler
- self.check_read()
-
- def check_read(self):
- self._stompConn._dispatcher.handle_read_event()
+ def add_client(self, sock):
+ adapter = StompAdapterImpl(self._reactor, self._sub_map,
+ self._req_dest)
+ return _StompConnection(self, adapter, sock,
+ self._reactor)
def send(
self,
@@ -185,21 +271,31 @@
destination=_DEFAULT_RESPONSE_DESTINATION,
):
self.log.debug("Sending response")
- res = stomp.Frame(
- stomp.Command.MESSAGE,
- {
- stomp.Headers.DESTINATION: destination,
- stomp.Headers.CONTENT_TYPE: "application/json",
- },
- message
- )
- self._stompConn.send_raw(res)
+
+ resp = json.loads(message)
+ try:
+ id = resp.get("id")
+ destination = self._req_dest[id]
+ del self._req_dest[id]
+ except:
+ # we could have no reply-to
+ pass
+
+ for connection in self._sub_map[destination]:
+ res = stomp.Frame(
+ stomp.Command.MESSAGE,
+ {
+ stomp.Headers.DESTINATION: destination,
+ stomp.Headers.CONTENT_TYPE: "application/json",
+ stomp.Headers.SUBSCRIPTION: connection.id
+ },
+ message
+ )
+ connection.client.send_raw(res)
def close(self):
- self._stompConn.close()
-
- def get_local_address(self):
- return self._socket.getsockname()[0]
+ for connection in self._sub_map.values():
+ connection.close()
class StompClient(object):
@@ -212,6 +308,7 @@
self._aclient = stomp.AsyncClient("vdsm")
self._stompConn = _StompConnection(
+ self,
self._aclient,
sock,
reactor
@@ -226,7 +323,7 @@
def handle_message(self, sub, frame):
if self._messageHandler is not None:
- self._messageHandler((self, frame.body))
+ self._messageHandler((self, self, frame.body))
def setMessageHandler(self, msgHandler):
self._messageHandler = msgHandler
@@ -241,7 +338,9 @@
*args,
**kwargs
):
- return self._aclient.subscribe(*args, **kwargs)
+ sub = self._aclient.subscribe(*args, **kwargs)
+ self._reactor.wakeup()
+ return sub
def send(
self,
@@ -264,8 +363,8 @@
self._stompConn.close()
-def StompListener(reactor, acceptHandler, connected_socket):
- impl = StompListenerImpl(reactor, acceptHandler, connected_socket)
+def StompListener(reactor, server, acceptHandler, connected_socket):
+ impl = StompListenerImpl(server, acceptHandler, connected_socket)
return Dispatcher(impl, connected_socket, map=reactor._map)
@@ -275,16 +374,16 @@
class StompListenerImpl(object):
log = logging.getLogger("jsonrpc.StompListener")
- def __init__(self, reactor, acceptHandler, connected_socket):
- self._reactor = reactor
+ def __init__(self, server, acceptHandler, connected_socket):
self._socket = connected_socket
self._acceptHandler = acceptHandler
+ self._server = server
def init(self, dispatcher):
dispatcher.set_reuse_addr()
- client = StompServer(self._socket, self._reactor)
- self._acceptHandler(self, client)
+ conn = self._server.add_client(self._socket)
+ self._acceptHandler(conn)
def writable(self, dispatcher):
return False
@@ -293,10 +392,12 @@
class StompReactor(object):
def __init__(self):
self._reactor = Reactor()
+ self._server = StompServer(self._reactor)
def createListener(self, connected_socket, acceptHandler):
listener = StompListener(
self._reactor,
+ self._server,
acceptHandler,
connected_socket
)
@@ -399,13 +500,9 @@
request_queue,
response_queue
):
- sub_id = None
- if request_queue == stomp.FAKE_SUB_ID:
- sub_id = stomp.FAKE_SUB_ID
-
return JsonRpcClient(
ClientRpcTransportAdapter(
- stomp_client.subscribe(response_queue, sub_id=sub_id),
+ stomp_client.subscribe(response_queue, sub_id=str(uuid4())),
request_queue,
stomp_client,
)
diff --git a/tests/jsonRpcHelper.py b/tests/jsonRpcHelper.py
index 2fb7cfb..ad9c00a 100644
--- a/tests/jsonRpcHelper.py
+++ b/tests/jsonRpcHelper.py
@@ -35,7 +35,8 @@
StompRpcClient,
)
from yajsonrpc.stomp import (
- FAKE_SUB_ID,
+ LEGACY_SUB_ID_REQ,
+ LEGACY_SUB_ID_RES,
)
from protocoldetector import MultiProtocolAcceptor
from rpc.BindingJsonRpc import BindingJsonRpc
@@ -122,8 +123,8 @@
def client(client_socket):
return StompRpcClient(
reactor.createClient(client_socket),
- FAKE_SUB_ID,
- FAKE_SUB_ID,
+ LEGACY_SUB_ID_REQ,
+ LEGACY_SUB_ID_RES,
)
def clientFactory():
diff --git a/vdsm/rpc/BindingJsonRpc.py b/vdsm/rpc/BindingJsonRpc.py
index eeea8ba..dc2d97e 100644
--- a/vdsm/rpc/BindingJsonRpc.py
+++ b/vdsm/rpc/BindingJsonRpc.py
@@ -41,7 +41,7 @@
def add_socket(self, reactor, client_socket):
reactor.createListener(client_socket, self._onAccept)
- def _onAccept(self, listener, client):
+ def _onAccept(self, client):
if self._clients is not None:
self._clients.append(client)
client.setMessageHandler(self._server.queueRequest)
--
To view, visit https://gerrit.ovirt.org/38451
To unsubscribe, visit https://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I1493070f2ba66ca9d39a6661876c82c4727cad62
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Piotr Kliczewski <piotr.kliczewski at gmail.com>
More information about the vdsm-patches
mailing list