code: def read_message(stream):
      
      answer = tornado.gen.Future()
      io_loop = IOLoop.current()
  
      def on_error(future):
          log.info('Failed to read data: %s', future.exception())
          return answer.set_exc_info(future.exc_info())
  
      @fail_to(answer)
      def on_body(size, future):
          if future.exception():
              return on_error(future)
  
          body = future.result()
          f = frame.frame_rw.read(BytesIO(body), size=size)
          message_type = f.header.message_type
          message_rw = messages.RW.get(message_type)
          if not message_rw:
              exc = errors.FatalProtocolError(
                  'Unknown message type %s', str(message_type)
              )
              return answer.set_exception(exc)
  
          message = message_rw.read(BytesIO(f.payload))
          message.id = f.header.message_id
          answer.set_result(message)
  
      @fail_to(answer)
      def on_read_size(future):
          if future.exception():
              return answer.set_exc_info(future.exc_info())
  
          size_bytes = future.result()
          size = frame.frame_rw.size_rw.read(BytesIO(size_bytes))
          io_loop.add_future(
              stream.read_bytes(size - FRAME_SIZE_WIDTH),
              lambda f: on_body(size, f)
          )
  
      try:
          # read_bytes may fail if the stream has already been closed
          read_size_future = stream.read_bytes(FRAME_SIZE_WIDTH)
      except Exception:
          answer.set_exc_info(sys.exc_info())
      else:
          read_size_future.add_done_callback(on_read_size)
      return answer
docs: Reads a message from the given IOStream.
  
      :param IOStream stream:
          IOStream to read from.