ThreadedZMQStream: close stream when it's done

not just the underlying socket,
because the stream represents event listeners on the socket,
and can produce errors or at least warnings if the socket is closed before being unregistered

jupyter_client/threaded.py

@@ -4,6 +4,8 @@
 import asyncio
 import atexit
 import time
+import warnings
+from concurrent.futures import Future
 from threading import Event, Thread
 from typing import Any, Dict, List, Optional
 
@@ -45,7 +47,7 @@ def __init__(
         session : :class:`session.Session`
             The session to use.
         loop
-            A pyzmq ioloop to connect the socket to using a ZMQStream
+            A tornado ioloop to connect the socket to using a ZMQStream
         """
         super().__init__()
 
@@ -80,6 +82,29 @@ def stop(self) -> None:
 
     def close(self) -> None:
         """ "Close the channel."""
+        if self.stream is not None and self.ioloop is not None:
+            # c.f.Future for threadsafe results
+            f = Future()
+
+            def close_stream():
+                try:
+                    self.stream.close()
+                    self.socket.close(linger=0)
+                    self.socket = None
+                except Exception as e:
+                    f.set_exception(e)
+                else:
+                    f.set_result(None)
+
+            self.ioloop.add_callback(close_stream)
+            try:
+                f.result(timeout=5)
+            except Exception as e:
+                # This should probably be log.warning, but we don't have self.log
+                warnings.warn(
+                    f"Error closing stream {self.stream}: {e}", RuntimeWarning, stacklevel=2
+                )
+
         if self.socket is not None:
             try:
                 self.socket.close(linger=0)