in the watcher: use the loop stored in the future

(this was breaking the tests)

docker/controller.py

@@ -799,8 +799,6 @@ class DockerWatcher:
         # {container_id: future}
         self._futures = {}
 
-        self._loop = asyncio.get_event_loop()
-
         self._shutdown = False
 
         self._thread = threading.Thread(target=self._thread_func)
@@ -816,22 +814,22 @@ class DockerWatcher:
         while not self._shutdown:   # pragma: nobranch
             next(limiter)
             try:
-                for event in self._client.events(filters={"event": "die"}):
+                for event_bytes in self._client.events(filters={"event": "die"}):
                     if self._shutdown:
                         return
-                    log.debug("docker event %r", event)
-                    self._loop.call_soon_threadsafe(self._event, event)
+
+                    log.debug("docker event %r", event_bytes)
+
+                    event = json.loads(event_bytes.decode())
+                    if event["status"] == "die":
+                        cid = event["id"]
+                        fut = self._futures.get(cid)
+                        if fut is not None and not fut.done():
+                            fut._loop.call_soon_threadsafe(fut.set_result, None)
             except Exception:
                 log.exception("docker watcher exception")
 
 
-    def _event(self, event_bytes):
-        event = json.loads(event_bytes.decode())
-        if event["status"] == "die":
-            cid = event["id"]
-            fut = self._futures.get(cid)
-            if fut is not None and not fut.done():
-                fut.set_result(None)
 
     def shutdown(self):
         """Shutdown the watcher