code: def _watch(self):
          async def _all_watcher():
              try:
                  allwatcher = client.AllWatcherFacade.from_connection(
                      self.connection())
                  while not self._watch_stopping.is_set():
                      try:
                          results = await utils.run_with_interrupt(
                              allwatcher.Next(),
                              self._watch_stopping,
                              loop=self._connector.loop)
                      except JujuAPIError as e:
                          if 'watcher was stopped' not in str(e):
                              raise
                          if self._watch_stopping.is_set():
                              break
                          log.warning(
                              'Watcher: watcher stopped, restarting')
                          del allwatcher.Id
                          continue
                      except websockets.ConnectionClosed:
                          monitor = self.connection().monitor
                          if monitor.status == monitor.ERROR:
                              log.warning(
                                  'Watcher: connection closed, reopening')
                              await self.connection().reconnect()
                              if monitor.status != monitor.CONNECTED:
                                  log.error('Watcher: automatic reconnect '
                                            'failed; stopping watcher')
                                  break
                              del allwatcher.Id
                              continue
                          else:
                              break
                      if self._watch_stopping.is_set():
                          try:
                              await allwatcher.Stop()
                          except websockets.ConnectionClosed:
                              pass
                          break
                      for delta in results.deltas:
                          try:
                              delta = get_entity_delta(delta)
                              old_obj, new_obj = self.state.apply_delta(delta)
                              await self._notify_observers(delta, old_obj, new_obj)
                          except KeyError as e:
                              log.debug("unknown delta type: %s", e.args[0])
                      self._watch_received.set()
              except CancelledError:
                  pass
              except Exception:
                  log.exception('Error in watcher')
                  raise
              finally:
                  self._watch_stopped.set()
          log.debug('Starting watcher task')
          self._watch_received.clear()
          self._watch_stopping.clear()
          self._watch_stopped.clear()
          self._connector.loop.create_task(_all_watcher())