PollScheduler: glib.io_add_watch() compatibility

diff --git a/pym/_emerge/AsynchronousLock.py b/pym/_emerge/AsynchronousLock.py
index 3593834f2c65e8a1a33e449df12745fc1a46c1d3..e166df344202ca7b98269a38cc8a57d54c22aa37 100644 (file)
--- a/pym/_emerge/AsynchronousLock.py
+++ b/pym/_emerge/AsynchronousLock.py
@@ -143,6 +143,8 @@ class _LockThread(AbstractPollTask):
                        self.returncode = os.EX_OK
                        self.wait()
 
+               return True
+
        def _cancel(self):
                # There's currently no way to force thread termination.
                pass
@@ -280,6 +282,8 @@ class _LockProcess(AbstractPollTask):
                        self.returncode = os.EX_OK
                        self.wait()
 
+               return True
+
        def _unregister(self):
                self._registered = False
 

diff --git a/pym/_emerge/EbuildIpcDaemon.py b/pym/_emerge/EbuildIpcDaemon.py
index 5dabe34b35f9b47edbf810cff79abbe7160a8c7b..6a320cb07197375abe770f8731b6c2a4ef1d18c4 100644 (file)
--- a/pym/_emerge/EbuildIpcDaemon.py
+++ b/pym/_emerge/EbuildIpcDaemon.py
@@ -84,6 +84,8 @@ class EbuildIpcDaemon(FifoIpcDaemon):
                                if reply_hook is not None:
                                        reply_hook()
 
+               return True
+
        def _send_reply(self, reply):
                # File streams are in unbuffered mode since we do atomic
                # read and write of whole pickles. Use non-blocking mode so

diff --git a/pym/_emerge/EbuildMetadataPhase.py b/pym/_emerge/EbuildMetadataPhase.py
index d4f5bc009ddfa61d74f48010e9b9397a7a3272b5..f8da8667f30828fe0b69bd0815b7abbae6f6697f 100644 (file)
--- a/pym/_emerge/EbuildMetadataPhase.py
+++ b/pym/_emerge/EbuildMetadataPhase.py
@@ -128,6 +128,8 @@ class EbuildMetadataPhase(SubProcess):
 
                self._unregister_if_appropriate(event)
 
+               return True
+
        def _set_returncode(self, wait_retval):
                SubProcess._set_returncode(self, wait_retval)
                # self._raw_metadata is None when _start returns

diff --git a/pym/_emerge/PipeReader.py b/pym/_emerge/PipeReader.py
index 0f784cf8bc75157ea71efd4f9a255fa2a2e95bf4..a85d794e21db9ad04a38157e4742318069fe298b 100644 (file)
--- a/pym/_emerge/PipeReader.py
+++ b/pym/_emerge/PipeReader.py
@@ -74,6 +74,8 @@ class PipeReader(AbstractPollTask):
 
                self._unregister_if_appropriate(event)
 
+               return True
+
        def _array_output_handler(self, fd, event):
 
                for f in self.input_files.values():
@@ -93,6 +95,8 @@ class PipeReader(AbstractPollTask):
 
                self._unregister_if_appropriate(event)
 
+               return True
+
        def _unregister(self):
                """
                Unregister from the scheduler and close open files.

diff --git a/pym/_emerge/PollScheduler.py b/pym/_emerge/PollScheduler.py
index 519a3700d6cccf14bffd44f13a3ad494f73d4997..7cf818957b12f04acc0d78c872a781e0c7ee3561 100644 (file)
--- a/pym/_emerge/PollScheduler.py
+++ b/pym/_emerge/PollScheduler.py
@@ -24,9 +24,12 @@ from _emerge.PollSelectAdapter import PollSelectAdapter
 class PollScheduler(object):
 
        class _sched_iface_class(SlotObject):
-               __slots__ = ("output", "register", "schedule",
+               __slots__ = ("io_add_watch", "output", "register", "schedule",
                        "source_remove", "timeout_add", "unregister")
 
+       class _io_handler_class(SlotObject):
+               __slots__ = ("args", "callback", "fd", "source_id")
+
        class _timeout_handler_class(SlotObject):
                __slots__ = ("args", "function", "interval", "source_id",
                        "timestamp")
@@ -49,6 +52,7 @@ class PollScheduler(object):
                self._scheduling = False
                self._background = False
                self.sched_iface = self._sched_iface_class(
+                       io_add_watch=self._register,
                        output=self._task_output,
                        register=self._register,
                        schedule=self._schedule_wait,
@@ -248,8 +252,9 @@ class PollScheduler(object):
                try:
                        while event_handlers:
                                f, event = self._next_poll_event()
-                               handler, reg_id = event_handlers[f]
-                               handler(f, event)
+                               x = event_handlers[f]
+                               if not x.callback(f, event, *x.args):
+                                       self._unregister(x.source_id)
                                event_handled = True
                except StopIteration:
                        event_handled = True
@@ -277,8 +282,9 @@ class PollScheduler(object):
                try:
                        while event_handlers and self._poll_event_queue:
                                f, event = self._next_poll_event()
-                               handler, reg_id = event_handlers[f]
-                               handler(f, event)
+                               x = event_handlers[f]
+                               if not x.callback(f, event, *x.args):
+                                       self._unregister(x.source_id)
                                events_handled += 1
                except StopIteration:
                        events_handled += 1
@@ -332,20 +338,31 @@ class PollScheduler(object):
 
                return bool(ready_timeouts)
 
-       def _register(self, f, eventmask, handler):
+       def _register(self, f, condition, callback, *args):
                """
-               @rtype: Integer
-               @return: A unique registration id, for use in schedule() or
-                       unregister() calls.
+               Like glib.io_add_watch(), your function should return False to
+               stop being called, or True to continue being called. Any
+               additional positional arguments given here are passed to your
+               function when it's called.
+
+               @type f: int or object with fileno() method
+               @param f: a file descriptor to monitor
+               @type condition: int
+               @param condition: a condition mask
+               @type callback: callable
+               @param callback: a function to call
+               @rtype: int
+               @return: an integer ID of the event source
                """
                if f in self._poll_event_handlers:
                        raise AssertionError("fd %d is already registered" % f)
                self._event_handler_id += 1
-               reg_id = self._event_handler_id
-               self._poll_event_handler_ids[reg_id] = f
-               self._poll_event_handlers[f] = (handler, reg_id)
-               self._poll_obj.register(f, eventmask)
-               return reg_id
+               source_id = self._event_handler_id
+               self._poll_event_handler_ids[source_id] = f
+               self._poll_event_handlers[f] = self._io_handler_class(
+                       args=args, callback=callback, f=f, source_id=source_id)
+               self._poll_obj.register(f, condition)
+               return source_id
 
        def _unregister(self, reg_id):
                """
@@ -409,8 +426,9 @@ class PollScheduler(object):
                        while (wait_ids is None and event_handlers) or \
                                (wait_ids is not None and wait_ids.intersection(handler_ids)):
                                f, event = self._next_poll_event(timeout=remaining_timeout)
-                               handler, reg_id = event_handlers[f]
-                               handler(f, event)
+                               x = event_handlers[f]
+                               if not x.callback(f, event, *x.args):
+                                       self._unregister(x.source_id)
                                event_handled = True
                                if condition is not None and condition():
                                        break

diff --git a/pym/_emerge/Scheduler.py b/pym/_emerge/Scheduler.py
index e6f3e0e5c4e840603aac0f9bdd822a2fcdc6b9e5..55e327f8fd1924a34d78e8f749f9994aea6a50cf 100644 (file)
--- a/pym/_emerge/Scheduler.py
+++ b/pym/_emerge/Scheduler.py
@@ -218,6 +218,7 @@ class Scheduler(PollScheduler):
                        schedule=self._schedule_fetch)
                self._sched_iface = self._iface_class(
                        fetch=fetch_iface, output=self._task_output,
+                       io_add_watch=self._register,
                        register=self._register,
                        schedule=self._schedule_wait,
                        scheduleSetup=self._schedule_setup,

diff --git a/pym/_emerge/SpawnProcess.py b/pym/_emerge/SpawnProcess.py
index ec5bf7d2f952bad23c0c84f2ef1df76359b7e296..411e7c66af6e0ab32d2e8733a997d995a620520a 100644 (file)
--- a/pym/_emerge/SpawnProcess.py
+++ b/pym/_emerge/SpawnProcess.py
@@ -218,6 +218,8 @@ class SpawnProcess(SubProcess):
 
                self._unregister_if_appropriate(event)
 
+               return True
+
        def _dummy_handler(self, fd, event):
                """
                This method is mainly interested in detecting EOF, since
@@ -240,6 +242,8 @@ class SpawnProcess(SubProcess):
 
                self._unregister_if_appropriate(event)
 
+               return True
+
        def _unregister(self):
                super(SpawnProcess, self)._unregister()
                if self._log_file_real is not None:

diff --git a/pym/portage/dbapi/_MergeProcess.py b/pym/portage/dbapi/_MergeProcess.py
index df501be385de988c0baac2eac8d7750ac107f535..eed7bd45f237e50a2e23f6047c3767899a1dde28 100644 (file)
--- a/pym/portage/dbapi/_MergeProcess.py
+++ b/pym/portage/dbapi/_MergeProcess.py
@@ -83,6 +83,8 @@ class MergeProcess(SpawnProcess):
                                        reporter = getattr(portage.elog.messages, funcname)
                                        reporter(msg, phase=phase, key=key, out=out)
 
+               return True
+
        def _spawn(self, args, fd_pipes, **kwargs):
                """
                Fork a subprocess, apply local settings, and call