distbuild: Move SubprocessEventSource into its own module

Previously it was only available in the distbuild-helper program. Moving
it to its own module means we can test it and reuse it.

This commit also adds a docstring to the class.

Change-Id: Iaf7854048cf0ff463a87894f1f500cdcb6a34d8b

4 files changed, 112 insertions, 75 deletions

diff --git a/distbuild-helper b/distbuild-helper
index d77fcaea..76a39d5f 100755
--- a/distbuild-helper
+++ b/distbuild-helper
@@ -32,78 +32,6 @@ import urlparse
 import distbuild
 
 
-class FileReadable(object):
-
-    def __init__(self, request_id, p, f):
-        self.request_id = request_id
-        self.process = p
-        self.file = f
-
-
-class FileWriteable(object):
-
-    def __init__(self, request_id, p, f):
-        self.request_id = request_id
-        self.process = p
-        self.file = f
-
-
-class SubprocessEventSource(distbuild.EventSource):
-
-    def __init__(self):
-        self.procs = []
-        self.closed = False
-
-    def get_select_params(self):
-        r = []
-        w = []
-        for requst_id, p in self.procs:
-            if p.stdin_contents is not None:
-                w.append(p.stdin)
-            if p.stdout is not None:
-                r.append(p.stdout)
-            if p.stderr is not None:
-                r.append(p.stderr)
-        return r, w, [], None
-
-    def get_events(self, r, w, x):
-        events = []
-
-        for request_id, p in self.procs:
-            if p.stdin in w:
-                events.append(FileWriteable(request_id, p, p.stdin))
-            if p.stdout in r:
-                events.append(FileReadable(request_id, p, p.stdout))
-            if p.stderr in r:
-                events.append(FileReadable(request_id, p, p.stderr))
-
-        return events
-
-    def add(self, request_id, process):
-
-        self.procs.append((request_id, process))
-        distbuild.set_nonblocking(process.stdin)
-        distbuild.set_nonblocking(process.stdout)
-        distbuild.set_nonblocking(process.stderr)
-
-    def remove(self, process):
-        self.procs = [t for t in self.procs if t[1] != process]
-
-    def kill_by_id(self, request_id):
-        logging.debug('SES: Killing all processes for %s', request_id)
-        for id, process in self.procs:
-            if id == request_id:
-                logging.debug('SES: killing %s', repr(process))
-                process.kill()
-
-    def close(self):
-        self.procs = []
-        self.closed = True
-
-    def is_finished(self):
-        return self.closed
-
-
 class HelperMachine(distbuild.StateMachine):
 
     def __init__(self, conn):
@@ -117,7 +45,7 @@ class HelperMachine(distbuild.StateMachine):
         jm = self.jm = distbuild.JsonMachine(self.conn)
         self.mainloop.add_state_machine(jm)
 
-        p = self.procsrc = SubprocessEventSource()
+        p = self.procsrc = distbuild.SubprocessEventSource()
         self.mainloop.add_event_source(p)
 
         self.send_helper_ready(jm)
@@ -125,8 +53,10 @@ class HelperMachine(distbuild.StateMachine):
         spec = [
             ('waiting', jm, distbuild.JsonNewMessage, 'waiting', self.do),
             ('waiting', jm, distbuild.JsonEof, None, self._eofed),
-            ('waiting', p, FileReadable, 'waiting', self._relay_exec_output),
-            ('waiting', p, FileWriteable, 'waiting', self._feed_stdin),
+            ('waiting', p, distbuild.FileReadable, 'waiting',
+             self._relay_exec_output),
+            ('waiting', p, distbuild.FileWriteable, 'waiting',
+             self._feed_stdin),
         ]
         self.add_transitions(spec)
 
diff --git a/distbuild/__init__.py b/distbuild/__init__.py
index 271b5def..e6ceda1f 100644
--- a/distbuild/__init__.py
+++ b/distbuild/__init__.py
@@ -62,4 +62,7 @@ from crashpoint import (crash_point, add_crash_condition, add_crash_conditions,
 
 from distbuild_socket import create_socket
 
+from subprocess_eventsrc import (FileReadable, FileWriteable,
+                                 SubprocessEventSource)
+
 __all__ = locals()
diff --git a/distbuild/subprocess_eventsrc.py b/distbuild/subprocess_eventsrc.py
new file mode 100644
index 00000000..52121502
--- /dev/null
+++ b/distbuild/subprocess_eventsrc.py
@@ -0,0 +1,103 @@
+# distbuild/subprocess_eventsrc.py -- for managing subprocesses
+#
+# Copyright (C) 2014-2015  Codethink Limited
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; version 2 of the License.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along
+# with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+
+import logging
+
+import distbuild
+
+
+class FileReadable(object):
+
+    def __init__(self, request_id, p, f):
+        self.request_id = request_id
+        self.process = p
+        self.file = f
+
+
+class FileWriteable(object):
+
+    def __init__(self, request_id, p, f):
+        self.request_id = request_id
+        self.process = p
+        self.file = f
+
+
+class SubprocessEventSource(distbuild.EventSource):
+    '''Event source for monitoring one or more subprocesses.
+
+    This will send FileReadable and FileWritable events based on the
+    stdin and stdout and stderr handles of each subprocesses.
+
+    When the subprocess terminates, you'll receive final FileReadable events
+    for stdout and for stderr. At that point, reading from those file
+    descriptors will return None, at which point you can be sure that the
+    subprocess is no longer running.
+
+    '''
+
+    def __init__(self):
+        self.procs = []
+        self.closed = False
+
+    def get_select_params(self):
+        r = []
+        w = []
+        for requst_id, p in self.procs:
+            if p.stdin_contents is not None:
+                w.append(p.stdin)
+            if p.stdout is not None:
+                r.append(p.stdout)
+            if p.stderr is not None:
+                r.append(p.stderr)
+        return r, w, [], None
+
+    def get_events(self, r, w, x):
+        events = []
+
+        for request_id, p in self.procs:
+            if p.stdin in w:
+                events.append(FileWriteable(request_id, p, p.stdin))
+            if p.stdout in r:
+                events.append(FileReadable(request_id, p, p.stdout))
+            if p.stderr in r:
+                events.append(FileReadable(request_id, p, p.stderr))
+
+        return events
+
+    def add(self, request_id, process):
+
+        self.procs.append((request_id, process))
+        distbuild.set_nonblocking(process.stdin)
+        distbuild.set_nonblocking(process.stdout)
+        distbuild.set_nonblocking(process.stderr)
+
+    def remove(self, process):
+        self.procs = [t for t in self.procs if t[1] != process]
+
+    def kill_by_id(self, request_id):
+        logging.debug('SES: Killing all processes for %s', request_id)
+        for id, process in self.procs:
+            if id == request_id:
+                logging.debug('SES: killing %s', repr(process))
+                process.kill()
+
+    def close(self):
+        self.procs = []
+        self.closed = True
+
+    def is_finished(self):
+        return self.closed
diff --git a/without-test-modules b/without-test-modules
index caf10c8b..fb66ff69 100644
--- a/without-test-modules
+++ b/without-test-modules
@@ -50,6 +50,7 @@ distbuild/proxy_event_source.py
 distbuild/sockbuf.py
 distbuild/socketsrc.py
 distbuild/sockserv.py
+distbuild/subprocess_eventsrc.py
 distbuild/timer_event_source.py
 distbuild/worker_build_scheduler.py
 # Not unit tested, since it needs a full system branch