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