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