Function ReadMessage

cpp/src/arrow/ipc/message.cc:403–449

}  // namespace

Result<std::unique_ptr<Message>> ReadMessage(std::shared_ptr<Buffer> metadata,
                                             std::shared_ptr<Buffer> body) {
  std::unique_ptr<Message> result;
  auto listener = std::make_shared<AssignMessageDecoderListener>(&result);
  // If the user does not pass in a body buffer then we assume they are skipping it
  MessageDecoder decoder(listener, default_memory_pool(), body == nullptr);

  if (metadata->size() < decoder.next_required_size()) {
    return Status::Invalid("metadata_length should be at least ",
                           decoder.next_required_size());
  }

  ARROW_RETURN_NOT_OK(decoder.Consume(metadata));
  if (decoder.buffered_size() > 0) {
    return Status::Invalid("Message metadata too long by ", decoder.buffered_size(),
                           " bytes");
  }

  switch (decoder.state()) {
    case MessageDecoder::State::INITIAL:
      // Metadata did not request a body so we better not have provided one
      DCHECK_EQ(body, nullptr);
      return result;
    case MessageDecoder::State::METADATA_LENGTH:
      return Status::Invalid("metadata length is missing from the metadata buffer");
    case MessageDecoder::State::METADATA:
      return Status::Invalid("flatbuffer size ", decoder.next_required_size(),
                             " invalid. Buffer size: ", metadata->size());
    case MessageDecoder::State::BODY: {
      if (body == nullptr) {
        // Caller didn't give a body so just give them a message without body
        return result;
      }
      if (body->size() != decoder.next_required_size()) {
        return Status::IOError("Expected body buffer to be ",
                               decoder.next_required_size(),
                               " bytes for message body, got ", body->size());
      }
      RETURN_NOT_OK(decoder.Consume(body));
      return result;
    }
    case MessageDecoder::State::EOS:
      return Status::Invalid("Unexpected empty message in IPC file format");
    default:
      return Status::Invalid("Unexpected state: ", decoder.state());
  }
}

// Common helper for the two ReadMessage overloads that take a file + offset.
// When body_length is provided, metadata and body are read in a single IO.