diff options
Diffstat (limited to 'zuul/driver/gerrit/gerritconnection.py')
| -rw-r--r-- | zuul/driver/gerrit/gerritconnection.py | 31 |
1 files changed, 13 insertions, 18 deletions
diff --git a/zuul/driver/gerrit/gerritconnection.py b/zuul/driver/gerrit/gerritconnection.py index 8502940c7..d8d90b125 100644 --- a/zuul/driver/gerrit/gerritconnection.py +++ b/zuul/driver/gerrit/gerritconnection.py @@ -67,9 +67,10 @@ class GerritChangeData(object): SSH = 1 HTTP = 2 - def __init__(self, fmt, data, related=None): + def __init__(self, fmt, data, related=None, files=None): self.format = fmt self.data = data + self.files = files if fmt == self.SSH: self.parseSSH(data) @@ -155,7 +156,7 @@ class GerritEventConnector(threading.Thread): time.sleep(1) def _run(self): - while not self._stopped: + while not self._stopped and self.event_queue.election.is_still_valid(): for event in self.event_queue: try: self._handleEvent(event) @@ -343,9 +344,6 @@ class GerritWatcher(threading.Thread): gerrit_connection.sched.zk_client, gerrit_connection.connection_name, "watcher") - self._connection_lost_event = threading.Event() - gerrit_connection.sched.zk_client.on_connection_lost_listeners.append( - self._connection_lost_event.set) self.keepalive = keepalive self._stopped = False @@ -363,9 +361,9 @@ class GerritWatcher(threading.Thread): def _listen(self, stdout, stderr): poll = select.poll() poll.register(stdout.channel) - while not (self._stopped or self._connection_lost_event.is_set()): + while not self._stopped and self.watcher_election.is_still_valid(): ret = poll.poll(self.poll_timeout) - if self._connection_lost_event.is_set(): + if not self.watcher_election.is_still_valid(): return for (fd, event) in ret: if fd == stdout.channel.fileno(): @@ -418,7 +416,6 @@ class GerritWatcher(threading.Thread): self.log.exception("Exception on ssh event stream with %s:", self.gerrit_connection.connection_name) self._stop_event.wait(5) - self._connection_lost_event.clear() def stop(self): self.log.debug("Stopping watcher") @@ -440,9 +437,6 @@ class GerritPoller(threading.Thread): connection.sched.zk_client, connection.connection_name, "poller") self._stopped = False self._stop_event = threading.Event() - self._connection_lost_event = threading.Event() - connection.sched.zk_client.on_connection_lost_listeners.append( - self._connection_lost_event.set) def _makePendingCheckEvent(self, change, uuid, check): return {'type': 'pending-check', @@ -497,7 +491,7 @@ class GerritPoller(threading.Thread): def _poll(self): next_start = self._last_start + self.poll_interval self._stop_event.wait(max(next_start - time.time(), 0)) - if self._stopped or self._connection_lost_event.is_set(): + if self._stopped or not self.poller_election.is_still_valid(): return self._last_start = time.time() self._poll_checkers() @@ -506,7 +500,7 @@ class GerritPoller(threading.Thread): def _run(self): self._last_start = time.time() - while not (self._stopped or self._connection_lost_event.is_set()): + while not self._stopped and self.poller_election.is_still_valid(): # during tests, a sub-class _poll method is used to send # notifications self._poll() @@ -519,7 +513,6 @@ class GerritPoller(threading.Thread): self.log.exception("Exception on Gerrit poll with %s:", self.connection.connection_name) time.sleep(1) - self._connection_lost_event.clear() def stop(self): self.log.debug("Stopping watcher") @@ -1247,16 +1240,18 @@ class GerritConnection(BaseConnection): 'o=DETAILED_LABELS' % (number,)) related = self.get('changes/%s/revisions/%s/related' % ( number, data['current_revision'])) - return data, related + files = self.get('changes/%s/revisions/%s/files?parent=1' % ( + number, data['current_revision'])) + return data, related, files def queryChange(self, number, event=None): for attempt in range(3): try: if self.session: - data, related = self.queryChangeHTTP(number, event=event) + data, related, files = self.queryChangeHTTP( + number, event=event) return GerritChangeData(GerritChangeData.HTTP, - data, - related) + data, related, files) else: data = self.queryChangeSSH(number, event=event) return GerritChangeData(GerritChangeData.SSH, data) |