Change in vdsm[master]: stomp: outgoing connection to a broker
piotr.kliczewski at gmail.com
piotr.kliczewski at gmail.com
Tue Mar 17 13:41:18 UTC 2015
Piotr Kliczewski has uploaded a new change for review.
Change subject: stomp: outgoing connection to a broker
......................................................................
stomp: outgoing connection to a broker
Change-Id: Ied3095a305fd5a5bfc19c8bc0342fba7cb811843
Signed-off-by: pkliczewski <piotr.kliczewski at gmail.com>
---
M lib/vdsm/config.py.in
M lib/yajsonrpc/__init__.py
M lib/yajsonrpc/stompReactor.py
M tests/jsonRpcHelper.py
M vdsm/clientIF.py
M vdsm/rpc/BindingJsonRpc.py
6 files changed, 109 insertions(+), 22 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/19/38819/1
diff --git a/lib/vdsm/config.py.in b/lib/vdsm/config.py.in
index cf8cadd..5689696 100644
--- a/lib/vdsm/config.py.in
+++ b/lib/vdsm/config.py.in
@@ -197,6 +197,8 @@
('jsonrpc_enable', 'true', 'Enable the JSON RPC server'),
+ ('broker_enable', 'false', 'Enable outgoing connection to broker'),
+
('report_host_threads_as_cores', 'false',
'Count each cpu hyperthread as an individual core'),
@@ -328,6 +330,19 @@
('guests_gateway_ip', '', None),
+ ('broker_address', '127.0.0.1',
+ 'Address where the broker is listening at. Use an empty string '
+ 'for none'),
+
+ ('broker_port', '5445',
+ 'Port where the broker is listening at.'),
+
+ ('request_queues',
+ 'jms.topic.vdsm_requests,jms.topic.vdsm_irs_requests',
+ 'Queues to subscribe to for RPC requests'),
+
+ ('event_queue', 'jms.topic.vdsm_requests',
+ 'Queue used for events'),
]),
# Section: [devel]
diff --git a/lib/yajsonrpc/__init__.py b/lib/yajsonrpc/__init__.py
index 48c3ca1..557f3d5 100644
--- a/lib/yajsonrpc/__init__.py
+++ b/lib/yajsonrpc/__init__.py
@@ -184,7 +184,6 @@
self._cb(
notification,
- self._event_id
)
diff --git a/lib/yajsonrpc/stompReactor.py b/lib/yajsonrpc/stompReactor.py
index c62869b..383ca49 100644
--- a/lib/yajsonrpc/stompReactor.py
+++ b/lib/yajsonrpc/stompReactor.py
@@ -20,17 +20,14 @@
import stomp
from vdsm import utils
+from vdsm.config import config
from vdsm.compat import json
from betterAsyncore import Dispatcher, Reactor
from vdsm.sslutils import SSLSocket
-from yajsonrpc import JsonRpcClient
+from yajsonrpc import JsonRpcClient, JsonRpcServer
_STATE_LEN = "Waiting for message length"
_STATE_MSG = "Waiting for message"
-
-
-_DEFAULT_RESPONSE_DESTINATION = "jms.topic.vdsm_legacy_responses"
-_DEFAULT_REQUEST_DESTINATION = "jms.topic.vdsm_legacy_requests"
def parseHeartBeatHeader(v):
@@ -61,6 +58,8 @@
self._sub_dests = sub_map
self._req_dest = req_dest
self._sub_ids = {}
+ request_queues = config.get('addresses', 'request_queues')
+ self.request_queues = request_queues.split(",")
self._commands = {
stomp.Command.CONNECT: self._cmd_connect,
stomp.Command.SEND: self._cmd_send,
@@ -163,7 +162,7 @@
def _cmd_send(self, dispatcher, frame):
destination = frame.headers.get(stomp.Headers.DESTINATION, None)
- if _DEFAULT_REQUEST_DESTINATION == destination:
+ if destination in self.request_queues:
# default subscription
self._handle_internal(dispatcher, stomp.Headers.REPLY_TO,
frame.body)
@@ -255,10 +254,10 @@
class StompServer(object):
log = logging.getLogger("yajsonrpc.StompServer")
- def __init__(self, reactor):
+ def __init__(self, reactor, subs):
self._reactor = reactor
self._messageHandler = None
- self._sub_map = {}
+ self._sub_map = subs
self._req_dest = {}
def add_client(self, sock):
@@ -270,7 +269,7 @@
def send(
self,
message,
- destination=_DEFAULT_RESPONSE_DESTINATION,
+ destination=stomp.LEGACY_SUB_ID_RES,
):
self.log.debug("Sending response")
try:
@@ -348,12 +347,9 @@
def send(
self,
message,
- destination=None,
+ destination=stomp.LEGACY_SUB_ID_RES,
headers=None
):
- if destination is None:
- destination = _DEFAULT_REQUEST_DESTINATION
-
self.log.debug("Sending request")
self._aclient.send(
destination,
@@ -393,9 +389,9 @@
class StompReactor(object):
- def __init__(self):
+ def __init__(self, subs):
self._reactor = Reactor()
- self._server = StompServer(self._reactor)
+ self._server = StompServer(self._reactor, subs)
def createListener(self, connected_socket, acceptHandler):
listener = StompListener(
@@ -440,6 +436,39 @@
self.log.debug("Stomp detected from %s", socket_address)
+class ServerRpcContextAdapter(object):
+ @classmethod
+ def subscription_handler(cls, server):
+ def handler(sub, frame):
+ server.queueRequest(
+ (
+ ServerRpcContextAdapter(sub.client, frame),
+ frame.body
+ )
+ )
+
+ return handler
+
+ def __init__(self, client, request_frame):
+ self._client = client
+ self._reply_to = request_frame.headers.get('reply-to', None)
+
+ def get_local_address(self, *args, **kwargs):
+ return ""
+
+ def send(self, data):
+ if self._reply_to is None:
+ return
+
+ self._client.send(
+ self._reply_to,
+ data,
+ {
+ "content-type": "application/json",
+ }
+ )
+
+
class ClientRpcTransportAdapter(object):
def __init__(self, sub, destination, client):
self._sub = sub
@@ -481,3 +510,18 @@
stomp_client,
)
)
+
+
+def StompRpcServer(
+ bridge,
+ stomp_client,
+ request_queue,
+):
+ server = JsonRpcServer(
+ bridge,
+ )
+
+ return stomp_client.subscribe(
+ request_queue,
+ message_handler=ServerRpcContextAdapter.subscription_handler(server)
+ )
diff --git a/tests/jsonRpcHelper.py b/tests/jsonRpcHelper.py
index 7fada3d..316fee4 100644
--- a/tests/jsonRpcHelper.py
+++ b/tests/jsonRpcHelper.py
@@ -95,7 +95,7 @@
xmlDetector = XmlDetector(xml_binding)
acceptor.add_detector(xmlDetector)
- json_binding = BindingJsonRpc(jsonBridge)
+ json_binding = BindingJsonRpc(jsonBridge, {})
json_binding.start()
cif.json_binding = json_binding
diff --git a/vdsm/clientIF.py b/vdsm/clientIF.py
index fe0f572a..dab9ac9 100644
--- a/vdsm/clientIF.py
+++ b/vdsm/clientIF.py
@@ -18,6 +18,7 @@
# Refer to the README and COPYING files for full details of the license
#
import os.path
+import socket
import time
import threading
import uuid
@@ -56,6 +57,10 @@
from yajsonrpc.betterAsyncore import (
Reactor,
)
+from yajsonrpc.stompReactor import (
+ StompClient,
+ StompRpcServer,
+)
from yajsonrpc import Notification
@@ -90,6 +95,8 @@
self._generationID = str(uuid.uuid4())
self.mom = None
self.bindings = {}
+ self._broker_client = None
+ self.subs = {}
if _glusterEnabled:
self.gluster = gapi.GlusterApi(self, log)
else:
@@ -112,6 +119,8 @@
host = config.get('addresses', 'management_ip')
port = config.getint('addresses', 'management_port')
+ self.event_destination = config.get('addresses', 'event_queue')
+
self._createReactor()
self._createAcceptor(host, port)
self._prepareXMLRPCBinding()
@@ -143,10 +152,10 @@
self._send_notification,
)
- def _send_notification(self, message, destination):
+ def _send_notification(self, message):
json_binding = self.bindings['jsonrpc']
server = json_binding.getStompReactor().server
- server.send(message)
+ server.send(message, self.event_destination)
def contEIOVms(self, sdUUID, isDomainStateValid):
# This method is called everytime the onDomainStateChange
@@ -194,6 +203,26 @@
sslctx
)
+ def _connectToBroker(self):
+ if config.getboolean('vars', 'broker_enable'):
+ broker_address = config.get('addresses', 'broker_address')
+ broker_port = config.getint('addresses', 'broker_port')
+ request_queues = config.get('addresses', 'request_queues')
+
+ sslctx = self._createSSLContext()
+ sock = socket.socket()
+ sock.connect((broker_address, broker_port))
+ if sslctx:
+ sock = sslctx.wrapSocket(sock)
+
+ self._broker_client = StompClient(sock, self._reactor)
+ for destination in request_queues.split(","):
+ self.subs[destination] = StompRpcServer(
+ self.bindings['jsonrpc'].server,
+ self._broker_client,
+ destination,
+ )
+
def _createSSLContext(self):
sslctx = None
if config.getboolean('vars', 'ssl'):
@@ -231,7 +260,7 @@
'Please make sure it is installed.')
else:
bridge = Bridge.DynamicBridge()
- json_binding = BindingJsonRpc(bridge)
+ json_binding = BindingJsonRpc(bridge, self.subs)
self.bindings['jsonrpc'] = json_binding
stomp_detector = StompDetector(json_binding)
self._acceptor.add_detector(stomp_detector)
diff --git a/vdsm/rpc/BindingJsonRpc.py b/vdsm/rpc/BindingJsonRpc.py
index c11a944..f766469 100644
--- a/vdsm/rpc/BindingJsonRpc.py
+++ b/vdsm/rpc/BindingJsonRpc.py
@@ -29,9 +29,9 @@
class BindingJsonRpc(object):
log = logging.getLogger('BindingJsonRpc')
- def __init__(self, bridge):
+ def __init__(self, bridge, subs):
self._server = JsonRpcServer(bridge, _simpleThreadFactory)
- self._reactor = StompReactor()
+ self._reactor = StompReactor(subs)
self.startReactor(self._reactor)
def add_socket(self, reactor, client_socket):
--
To view, visit https://gerrit.ovirt.org/38819
To unsubscribe, visit https://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ied3095a305fd5a5bfc19c8bc0342fba7cb811843
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