Re-add _RietveldPoller

scripts/master/try_job_rietveld.py

 # Copyright (c) 2012 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 import datetime
 import hashlib
 import json
 import os
 import pytz
 import time
@@ skipping 99 matching lines @@
 
   def _MakeSet(self, data):
     """Converts the input data string into a set of email addresses.
     """
     emails = (email.strip() for email in data.splitlines())
     self._users = frozenset(email for email in emails if email)
     log.msg('Found %d users' % len(self._users))
     self._valid = True
 
 
+class _RietveldPoller(base.PollingChangeSource):
+  """Polls Rietveld for any pending patch sets to build.
+
+  Periodically polls Rietveld to see if any patch sets have been marked by
+  users to be tried.  If so, send them to the trybots.
+  """
+
+  def __init__(self, get_pending_endpoint, interval, cachepath=None):
+    """
+    Args:
+      get_pending_endpoint: Rietveld URL string used to retrieve jobs to try.
+      interval: Interval used to poll Rietveld, in seconds.
+      cachepath: Path to file where state to persist between master restarts
+                 will be stored.
+    """
+    # Set interval time in base class.
+    self.pollInterval = interval
+
+    # A string URL for the Rietveld endpoint to query for pending try jobs.
+    self._get_pending_endpoint = get_pending_endpoint
+
+    # Cursor used to keep track of next patchset(s) to try.  If the cursor
+    # is None, then try from the beginning.
+    self._cursor = None
+
+    # Try job parent of this poller.
+    self._try_job_rietveld = None
+
+    self._cachepath = cachepath
+
+    if self._cachepath:
+      if os.path.exists(self._cachepath):
+        with open(self._cachepath) as f:
+          # Using JSON allows us to be flexible and extend the format
+          # in a compatible way.
+          data = json.load(f)
+          self._cursor = data.get('cursor').encode('utf-8')
+
+  # base.PollingChangeSource overrides:
+  def poll(self):
+    """Polls Rietveld for any pending try jobs and submit them.
+
+    Returns:
+      A deferred objects to be called once the operation completes.
+    """
+    log.msg('RietveldPoller.poll')
+    d = defer.succeed(None)
+    d.addCallback(self._OpenUrl)
+    d.addCallback(self._ParseJson)
+    d.addErrback(log.err, 'error in RietveldPoller')  # eat errors
+    return d
+
+  def setServiceParent(self, parent):
+    base.PollingChangeSource.setServiceParent(self, parent)
+    self._try_job_rietveld = parent
+
+  def _OpenUrl(self, _):
+    """Downloads pending patch sets from Rietveld.
+
+    Returns: A string containing the pending patchsets from Rietveld
+        encoded as JSON.
+    """
+    endpoint = self._get_pending_endpoint
+    if self._cursor:
+      sep = '&' if '?' in endpoint else '?'
+      endpoint = endpoint + '%scursor=%s' % (sep, self._cursor)
+
+    log.msg('RietveldPoller._OpenUrl: %s' % endpoint)
+    return client.getPage(endpoint, agent='buildbot', timeout=2*60)
+
+  def _ParseJson(self, json_string):
+    """Parses the JSON pending patch set information.
+
+    Args:
+      json_string: A string containing the serialized JSON jobs.
+
+    Returns: A list of pending try jobs.  This is the list of all jobs returned
+        by Rietveld, not simply the ones we tried this time.
+    """
+    data = json.loads(json_string)
+    d = self._try_job_rietveld.SubmitJobs(data['jobs'])
+    def success_callback(value):
+      self._cursor = str(data['cursor'])
+      self._try_job_rietveld.processed_keys.clear()
+
+      if self._cachepath:
+        with open(self._cachepath, 'w') as f:
+          json.dump({'cursor': self._cursor}, f)
+
+      return value
+    d.addCallback(success_callback)
+    return d
+
+
 class _RietveldPollerWithCache(base.PollingChangeSource):
   """Polls Rietveld for any pending patch sets to build.
 
   Periodically polls Rietveld to see if any patch sets have been marked by
   users to be tried. If so, send them to the trybots. Uses cursor to download
   all pages within a single poll. To avoid sending same jobs, keeps a cache of
   the jobs that were already processed thus reducing chances of a duplicate job
   submission and increasing robustness to bugs in Rietveld. On the first poll
   this cache is initialized with jobs currently pending on the Buildbot.
   """
@@ skipping 296 matching lines @@
         else:
           self.processed_keys.add(job['key'])
           log.err('Rietveld not updated: no corresponding service found.')
 
   # TryJobBase overrides:
   def setServiceParent(self, parent):
     TryJobBase.setServiceParent(self, parent)
     self._poller.setServiceParent(self)
     self._poller.master = self.master
     self._valid_users.setServiceParent(self)