Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[v8.0] fix (StompMQConnector): add a timeout for the StompConnector #7854

Open
wants to merge 2 commits into
base: rel-v8r0
Choose a base branch
from
Open

[v8.0] fix (StompMQConnector): add a timeout for the StompConnector #7854

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f3df507
fix (StompMQConnector): add a timeout for the StompConnector
ryuwd Oct 23, 2024
612065e
fix (StompMQConnector): add heartbeats parameter to StompConnector
ryuwd Oct 24, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions src/DIRAC/Resources/MessageQueue/StompMQConnector.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,10 @@ class StompMQConnector(MQConnector):
RECONNECT_SLEEP_JITTER = 0.1 # Random factor to add. 0.1 means a random number from 0 to 10% of the current time.
RECONNECT_ATTEMPTS_MAX = 1e4 # Maximum attempts to reconnect.

OUTGOING_HEARTBEAT_MS = 15_000
INCOMING_HEARTBEAT_MS = 15_000
STOMP_TIMEOUT = 60

PORT = 61613

def __init__(self, parameters=None):
Expand Down Expand Up @@ -72,6 +76,11 @@ def setupConnection(self, parameters=None):
reconnectSleepJitter = self.parameters.get("ReconnectSleepJitter", StompMQConnector.RECONNECT_SLEEP_JITTER)
reconnectAttemptsMax = self.parameters.get("ReconnectAttemptsMax", StompMQConnector.RECONNECT_ATTEMPTS_MAX)

outgoingHeartbeatMs = self.parameters.get("OutgoingHeartbeatMs", StompMQConnector.OUTGOING_HEARTBEAT_MS)
incomingHeartbeatMs = self.parameters.get("IncomingHeartbeatMs", StompMQConnector.INCOMING_HEARTBEAT_MS)

stompTimeout = self.parameters.get("Timeout", StompMQConnector.STOMP_TIMEOUT)

host = self.parameters.get("Host")
port = self.parameters.get("Port", StompMQConnector.PORT)
vhost = self.parameters.get("VHost")
Expand All @@ -83,6 +92,8 @@ def setupConnection(self, parameters=None):
connectionArgs = {
"vhost": vhost,
"keepalive": True,
"timeout": stompTimeout,
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Having tried it we also need something like "heartbeats": (outgoingHeartbeatMs, incomingHeartbeatMs), here as well. 15_000 seems like a sensible default

"heartbeats": (outgoingHeartbeatMs, incomingHeartbeatMs),
"reconnect_sleep_initial": reconnectSleepInitial,
"reconnect_sleep_increase": reconnectSleepIncrease,
"reconnect_sleep_max": reconnectSleepMax,
Expand Down
Loading