Adds set_close_callback function to removes stream instance after closed from a set streams.

salt/transport/ipc.py

@@ -559,6 +559,11 @@ class IPCMessagePublisher(object):
                     io_loop=self.io_loop
                 )
             self.streams.add(stream)
+
+            def discard_after_closed():
+                self.streams.discard(stream)
+
+            stream.set_close_callback(discard_after_closed)
         except Exception as exc:
             log.error('IPC streaming error: {0}'.format(exc))