code: def _handle_message(self, opcode: int, data: bytes) -> "Optional[Future[None]]":
          
          if self.client_terminated:
              return None
  
          if self._frame_compressed:
              assert self._decompressor is not None
              try:
                  data = self._decompressor.decompress(data)
              except _DecompressTooLargeError:
                  self.close(1009, "message too big after decompression")
                  self._abort()
                  return None
  
          if opcode == 0x1:
              # UTF-8 data
              self._message_bytes_in += len(data)
              try:
                  decoded = data.decode("utf-8")
              except UnicodeDecodeError:
                  self._abort()
                  return None
              return self._run_callback(self.handler.on_message, decoded)
          elif opcode == 0x2:
              # Binary data
              self._message_bytes_in += len(data)
              return self._run_callback(self.handler.on_message, data)
          elif opcode == 0x8:
              # Close
              self.client_terminated = True
              if len(data) >= 2:
                  self.close_code = struct.unpack(">H", data[:2])[0]
              if len(data) > 2:
                  self.close_reason = to_unicode(data[2:])
              # Echo the received close code, if any (RFC 6455 section 5.5.1).
              self.close(self.close_code)
          elif opcode == 0x9:
              # Ping
              try:
                  self._write_frame(True, 0xA, data)
              except StreamClosedError:
                  self._abort()
              self._run_callback(self.handler.on_ping, data)
          elif opcode == 0xA:
              # Pong
              self.last_pong = IOLoop.current().time()
              return self._run_callback(self.handler.on_pong, data)
          else:
              self._abort()
          return None
docs: Execute on_message, returning its Future if it is a coroutine.