saltstack/salt
1
0
Fork 0
mirror of https://github.com/saltstack/salt.git synced 2025-04-17 10:10:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Revert change to zmq transport

Browse source
This commit is contained in:
Daniel A. Wozniak 2023-11-24 02:31:50 -07:00 • committed by Daniel Wozniak
parent 10b714cf55
commit e39246b1ef
1 changed files with 55 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

72
salt/transport/zeromq.py
View file

@ -342,28 +342,66 @@ class PublishClient(salt.transport.base.PublishClient):
# raise Exception("Send not supported")
# await self._socket.send(msg)
async def on_recv_handler(self, callback):
while not self._socket:
# Retry quickly, we may want to increase this if it's hogging cpu.
await asyncio.sleep(0.003)
while True:
msg = await self.recv()
if msg:
await callback(msg)
#async def on_recv_handler(self, callback):
# while not self._socket:
# # Retry quickly, we may want to increase this if it's hogging cpu.
# await asyncio.sleep(0.003)
# while True:
# msg = await self.recv()
# if msg:
# await callback(msg)
#def on_recv(self, callback):
# """
# Register a callback for received messages (that we didn't initiate)
# """
# if self.on_recv_task:
# # XXX: We are not awaiting this canceled task. This still needs to
# # be addressed.
# self.on_recv_task.cancel()
# if callback is None:
# self.on_recv_task = None
# else:
# self.on_recv_task = asyncio.create_task(self.on_recv_handler(callback))
def on_recv(self, callback):
"""
Register a callback for received messages (that we didn't initiate)
"""
if self.on_recv_task:
# XXX: We are not awaiting this canceled task. This still needs to
# be addressed.
self.on_recv_task.cancel()
if callback is None:
self.on_recv_task = None
else:
self.on_recv_task = asyncio.create_task(self.on_recv_handler(callback))
:param func callback: A function which should be called when data is received
"""
if callback is None:
callbacks = self.callbacks
self.callbacks = {}
for callback, (running, task) in callbacks.items():
running.clear()
return
running = asyncio.Event()
running.set()
async def consume(running):
try:
while running.is_set():
try:
msg = await self.recv(timeout=None)
except zmq.error.ZMQError as exc:
# We've disconnected just die
break
if msg:
try:
await callback(msg)
except Exception: # pylint: disable=broad-except
log.error("Exception while running callback", exc_info=True)
# log.debug("Callback done %r", callback)
except Exception as exc: # pylint: disable=broad-except
log.error(
"Exception while consuming%s %s", self.uri, exc, exc_info=True
)
task = self.io_loop.spawn_callback(consume, running)
self.callbacks[callback] = running, task
class RequestServer(salt.transport.base.DaemonizedRequestServer):