Change in vdsm[master]: stomp: make sure that stomp connection is active before send...

piotr.kliczewski at gmail.com piotr.kliczewski at gmail.com
Mon Nov 23 13:37:12 UTC 2015


Piotr Kliczewski has uploaded a new change for review.

Change subject: stomp: make sure that stomp connection is active before sending messages
......................................................................

stomp: make sure that stomp connection is active before sending messages

In may happen that client sends SEND frame before CONNECT and SUBSCRIBE
frames are processed by the server which can lead to server having no
knowledge to whom send response because subscription is not finalized
yet. We need to wait for server to respond with CONNECTED frame before
sending any rpc requests.


Change-Id: I4f6881d7966fa47031d027740633ca83c834387f
Signed-off-by: pkliczewski <piotr.kliczewski at gmail.com>
---
M lib/yajsonrpc/stomp.py
M tests/stompAsyncClientTests.py
2 files changed, 14 insertions(+), 0 deletions(-)


  git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/00/49000/1

diff --git a/lib/yajsonrpc/stomp.py b/lib/yajsonrpc/stomp.py
index 38c05e4..b02039b 100644
--- a/lib/yajsonrpc/stomp.py
+++ b/lib/yajsonrpc/stomp.py
@@ -18,6 +18,8 @@
 import socket
 from uuid import uuid4
 from collections import deque
+from threading import Event
+from . import CALL_TIMEOUT
 
 from vdsm import utils
 import re
@@ -434,6 +436,7 @@
         self._outbox = deque()
         self._error = None
         self._subscriptions = {}
+        self._ready = Event()
         self._commands = {
             Command.CONNECTED: self._process_connected,
             Command.MESSAGE: self._process_message,
@@ -481,6 +484,7 @@
         self._connected = True
 
         self.log.debug("Stomp connection established")
+        self._ready.set()
 
     def _process_message(self, frame, dispatcher):
         sub_id = frame.headers.get(Headers.SUBSCRIPTION)
@@ -504,6 +508,9 @@
         raise StompError(frame)
 
     def send(self, destination, data="", headers=None):
+        if not self._ready.wait(timeout=CALL_TIMEOUT):
+            raise TimeoutError()
+
         final_headers = {"destination": destination}
         if headers is not None:
             final_headers.update(headers)
@@ -591,3 +598,7 @@
     def unsubscribe(self):
         self._client.unsubscribe(self)
         self._valid = False
+
+
+class TimeoutError(Exception):
+    pass
diff --git a/tests/stompAsyncClientTests.py b/tests/stompAsyncClientTests.py
index 4ffc327..8d10a23 100644
--- a/tests/stompAsyncClientTests.py
+++ b/tests/stompAsyncClientTests.py
@@ -80,6 +80,9 @@
                 '"id":"e8a936a6-d886-4cfa-97b9-2d54209053ff"}')
         headers = {Headers.REPLY_TO: 'jms.topic.vdsm_responses',
                    Headers.CONTENT_LENGTH: '103'}
+        # make sure that client can send messages
+        client._ready.set()
+
         client.send('jms.topic.vdsm_requests', data, headers)
 
         req_frame = client.pop_message()


-- 
To view, visit https://gerrit.ovirt.org/49000
To unsubscribe, visit https://gerrit.ovirt.org/settings

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