Re-add _RietveldPoller

scripts/master/try_job_rietveld.py

Index: scripts/master/try_job_rietveld.py
diff --git a/scripts/master/try_job_rietveld.py b/scripts/master/try_job_rietveld.py
index a351ad38e4f57afbd620b163be61c139d926a251..983c2e5979670d8b26867ea7cf16b1fc2be1e19d 100644
--- a/scripts/master/try_job_rietveld.py
+++ b/scripts/master/try_job_rietveld.py
@@ -117,6 +117,100 @@ class _ValidUserPoller(internet.TimerService):
     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.