Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def add_reader(self, fd, callback, *args):
if not isinstance(fd, int):
fd = fd.fileno()
self.remove_reader(fd)
s = GLib.unix_fd_source_new(fd, GLib.IO_IN)
assert fd not in self._readers
self._readers[fd] = GLibHandle(self, s, True, callback, args)
def add_writer(self, fd, callback, *args):
if not isinstance(fd, int):
fd = fd.fileno()
self.remove_writer(fd)
s = GLib.unix_fd_source_new(fd, GLib.IO_OUT)
assert fd not in self._writers
self._writers[fd] = GLibHandle(self, s, True, callback, args)
def call_later(self, delay, callback, *args):
if delay <= 0:
return self.call_soon(callback, *args)
else:
return GLibHandle(
self,
GLib.Timeout(delay*1000 if delay > 0 else 0),
False,
callback, args)
def add_signal_handler(self, sig, callback, *args):
self._check_signal(sig)
self.remove_signal_handler(sig)
s = GLib.unix_signal_source_new(sig)
if s is None:
if sig == signal.SIGKILL:
raise RuntimeError("cannot catch SIGKILL")
else:
raise ValueError("signal not supported")
assert sig not in self._sighandlers
self._sighandlers[sig] = GLibHandle(self, s, True, callback, args)