Add LineConsumerLogObserver

master/buildbot/process/logobserver.py

@@ -106,6 +106,37 @@ def headerLineReceived(self, line):
         pass
 
 
+class LineConsumerLogObserver(LogLineObserver):
+
+    def __init__(self, consumerFunction):
+        LogLineObserver.__init__(self)
+        self.generator = None
+        self.consumerFunction = consumerFunction
+
+    def feed(self, input):
+        # note that we defer starting the generator until the first bit of
+        # data, since the observer will be instantiated during configuration
+        # as well as during
+        self.generator = self.consumerFunction()
+        self.generator.next()
+        # shortcut all remaining feed operations
+        self.feed = self.generator.send
+        self.feed(input)
+
+    def outLineReceived(self, line):
+        self.feed(('o', line))
+
+    def errLineReceived(self, line):
+        self.feed(('e', line))
+
+    def headerLineReceived(self, line):
+        self.feed(('h', line))
+
+    def finishReceived(self):
+        if self.generator:
+            self.generator.close()
+
+
 class OutputProgressObserver(LogObserver):
     length = 0
 

master/buildbot/test/fake/logfile.py

@@ -101,6 +101,6 @@ def fakeData(self, header='', stdout='', stderr=''):
     # removed methods, here temporarily
 
     def getText(self):
-        warnings.warn("step uses removed LogFile method `getText`")
+        warnings.warn("step uses removed LogFile method `getText`", stacklevel=2)
         return ''.join([c for str, c in self.chunks
                         if str in (STDOUT, STDERR)])

master/buildbot/test/unit/test_process_logobserver.py

@@ -88,6 +88,70 @@ def finishReceived(self):
         self.obs.append(('fin',))
 
 
+class TestLineConsumerLogObesrver(unittest.TestCase):
+
+    def setUp(self):
+        self.master = fakemaster.make_master(testcase=self, wantData=True)
+
+    @defer.inlineCallbacks
+    def do_test_sequence(self, consumer):
+        logid = yield self.master.data.updates.newLog(1, u'mine', u's')
+        l = log.Log.new(self.master, 'mine', 's', logid, 'utf-8')
+        lo = logobserver.LineConsumerLogObserver(consumer)
+        lo.setLog(l)
+
+        yield l.addStdout(u'hello\n')
+        yield l.addStderr(u'cruel\n')
+        yield l.addStdout(u'multi\nline\nchunk\n')
+        yield l.addHeader(u'H1\nH2\n')
+        yield l.finish()
+
+    @defer.inlineCallbacks
+    def test_sequence_finish(self):
+        results = []
+
+        def consumer():
+            while True:
+                try:
+                    stream, line = yield
+                    results.append((stream, line))
+                except GeneratorExit:
+                    results.append('finish')
+                    raise
+        yield self.do_test_sequence(consumer)
+
+        self.assertEqual(results, [
+            ('o', u'hello'),
+            ('e', u'cruel'),
+            ('o', u'multi'),
+            ('o', u'line'),
+            ('o', u'chunk'),
+            ('h', u'H1'),
+            ('h', u'H2'),
+            'finish',
+        ])
+
+    @defer.inlineCallbacks
+    def test_sequence_no_finish(self):
+        results = []
+
+        def consumer():
+            while True:
+                stream, line = yield
+                results.append((stream, line))
+        yield self.do_test_sequence(consumer)
+
+        self.assertEqual(results, [
+            ('o', u'hello'),
+            ('e', u'cruel'),
+            ('o', u'multi'),
+            ('o', u'line'),
+            ('o', u'chunk'),
+            ('h', u'H1'),
+            ('h', u'H2'),
+        ])
+
+
 class TestLogLineObserver(unittest.TestCase):
 
     def setUp(self):

master/docs/developer/cls-logobserver.rst

@@ -66,6 +66,42 @@ LogObservers
 
         This method, inherited from :py:class:`LogObserver`, is invoked when the observed log is finished.
 
+.. py:class:: LineConsumerLogObserver
+
+    This subclass of :py:class:`LogObserver` takes a generator function and "sends" each line to that function.
+    This allows consumers to be written as stateful Python functions, e.g., ::
+
+        def logConsumer(self):
+            while True:
+                stream, line = yield
+                if stream == 'o' and line.startswith('W'):
+                    self.warnings.append(line[1:])
+
+        def __init__(self):
+            ...
+            self.warnings = []
+            self.addLogObserver('stdio', logobserver.LineConsumerLogObserver(self.logConsumer))
+
+    Each ``yield`` expression evaluates to a tuple of (stream, line), where the stream is one of 'o', 'e', or 'h' for stdout, stderr, and header, respectively.
+    As with any generator function, the ``yield`` expression will raise a ``GeneratorExit`` exception when the generator is complete.
+    To do something after the log is finished, just catch this exception (but then re-raise it or return) ::
+
+        def logConsumer(self):
+            while True:
+                try:
+                    stream, line = yield
+                    if stream == 'o' and line.startswith('W'):
+                        self.warnings.append(line[1:])
+                except GeneratorExit:
+                    self.warnings.sort()
+                    return
+
+    .. warning::
+
+        This use of generator functions is a simple Python idiom first described in [PEP 342](http://www.python.org/dev/peps/pep-0342/).
+        It is unrelated to the generators used in ``inlineCallbacks``.
+        In fact, consumers of this type are incompatible with asynchronous programming, as each line must be processed immediately.
+
 .. py:class:: BufferLogObserver(wantStdout=True, wantStderr=False)
 
     :param boolean wantStdout: true if stdout should be buffered