code: def handle_stream(self, stream, address):
          '''
          Handle incoming streams and add messages to the incoming queue
          '''
          log.trace('Req client %s connected', address)
          self.clients.append((stream, address))
          unpacker = msgpack.Unpacker()
          try:
              while True:
                  wire_bytes = yield stream.read_bytes(4096, partial=True)
                  unpacker.feed(wire_bytes)
                  for framed_msg in unpacker:
                      if six.PY3:
                          framed_msg = salt.transport.frame.decode_embedded_strs(
                              framed_msg
                          )
                      header = framed_msg['head']
                      self.io_loop.spawn_callback(self.message_handler, stream, header, framed_msg['body'])
  
          except StreamClosedError:
              log.trace('req client disconnected %s', address)
              self.clients.remove((stream, address))
          except Exception as e:
              log.trace('other master-side exception: %s', e)
              self.clients.remove((stream, address))
              stream.close(f handle_stream(self, stream, address):
          '''
          Handle incoming streams and add messages to the incoming queue
          '''
          log.trace('Req client %s connected', address)
          self.clients.append((stream, address))
          unpacker = msgpack.Unpacker()
          try:
              while True:
                  wire_bytes = yield stream.read_bytes(4096, partial=True)
                  unpacker.feed(wire_bytes)
                  for framed_msg in unpacker:
                      if six.PY3:
                          framed_msg = salt.transport.frame.decode_embedded_strs(
                              framed_msg
                          )
                      header = framed_msg['head']
                      self.io_loop.spawn_callback(self.message_handler, stream, header, framed_msg['body'])
  
          except StreamClosedError:
              log.trace('req client disconnected %s', address)
              self.clients.remove((stream, address))
          except Exception as e:
              log.trace('other master-side exception: %s', e)
              self.clients.remove((stream, address))
              stream.close()
docs: Handle incoming streams and add messages to the incoming queue