def _notification_worker(self):
    while self._running:
      work = self._notification_queue.get()
      self._cb_blocker.ensure_safe()
      try:
        new_nodes, removed_nodes = work
        new_members = self._zk_nodes_to_members(new_nodes)
        self._members.update(((m.name, m) for m in new_members))

        self._log.debug("Raising notifications for %i members joining and %i members leaving."
                 % (len(new_nodes), len(removed_nodes)))

        for m in removed_nodes:
          removed_member = self._members.pop(m, None)
          if removed_member:
            try:
              self._on_leave(removed_member)
            except Exception:
              self._log.exception('Error in OnLeave callback.')
          else:
            self._log.warn('Member %s was not found in cached set' % str(m))

        for m in new_members:
          try:
            self._on_join(m)
          except Exception:
            self._log.exception('Error in OnJoin callback.')
      except Exception:
        self._log.exception('Error in notification worker.')
'Atomically' raise notifications for join / leave.
dataset/ETHPy150Open steveniemitz/scales/scales/loadbalancer/zookeeper.py/ServerSet._notification_worker