webhooks: Use reactor to watch for writable fds

Signed-off-by: Kevin O'Connor <kevin@koconnor.net>
This commit is contained in:
Kevin O'Connor 2022-05-23 21:57:04 -04:00
parent 02dd0742c4
commit 7b9583391e
1 changed files with 22 additions and 27 deletions

View File

@ -171,9 +171,9 @@ class ClientConnection:
self.uid = id(self)
self.sock = sock
self.fd_handle = self.reactor.register_fd(
self.sock.fileno(), self.process_received)
self.sock.fileno(), self.process_received, self._do_send)
self.partial_data = self.send_buffer = b""
self.is_sending_data = False
self.is_blocking = False
self.set_client_info("?", "New connection")
self.request_log = collections.deque([], REQUEST_LOG_SIZE)
@ -259,33 +259,28 @@ class ClientConnection:
def send(self, data):
jmsg = json.dumps(data, separators=(',', ':'))
self.send_buffer += jmsg.encode() + b"\x03"
if not self.is_sending_data:
self.is_sending_data = True
self.reactor.register_callback(self._do_send)
if not self.is_blocking:
self._do_send()
def _do_send(self, eventtime):
retries = 10
while self.send_buffer:
def _do_send(self, eventtime=None):
if self.fd_handle is None:
return
try:
sent = self.sock.send(self.send_buffer)
except socket.error as e:
if e.errno == errno.EBADF or e.errno == errno.EPIPE \
or not retries:
sent = 0
else:
retries -= 1
waketime = self.reactor.monotonic() + .001
self.reactor.pause(waketime)
continue
retries = 10
if sent > 0:
self.send_buffer = self.send_buffer[sent:]
else:
logging.info(
"webhooks: Error sending server data, closing socket")
if e.errno not in [errno.EAGAIN, errno.EWOULDBLOCK]:
logging.info("webhooks: socket write error %d" % (self.uid,))
self.close()
break
self.is_sending_data = False
return
sent = 0
if sent < len(self.send_buffer):
if not self.is_blocking:
self.reactor.set_fd_wake(self.fd_handle, False, True)
self.is_blocking = True
elif self.is_blocking:
self.reactor.set_fd_wake(self.fd_handle, True, False)
self.is_blocking = False
self.send_buffer = self.send_buffer[sent:]
class WebHooks:
def __init__(self, printer):