(self)
| 159 | self._tref = self.timer.call_repeatedly(self.interval, self.periodic) |
| 160 | |
| 161 | def periodic(self): |
| 162 | workers = self.state.workers |
| 163 | dirty = set() |
| 164 | for worker in workers.values(): |
| 165 | if not worker.alive: |
| 166 | dirty.add(worker) |
| 167 | self.on_node_lost(worker) |
| 168 | for worker in dirty: |
| 169 | workers.pop(worker.hostname, None) |
| 170 | |
| 171 | def get_consumers(self, channel): |
| 172 | self.register_timer() |