Function ReadMessageAsync

cpp/src/arrow/ipc/message.cc:458–510

}

Future<std::shared_ptr<Message>> ReadMessageAsync(int64_t offset, int32_t metadata_length,
                                                  int64_t body_length,
                                                  io::RandomAccessFile* file,
                                                  const io::IOContext& context) {
  struct State {
    std::unique_ptr<Message> result;
    std::shared_ptr<MessageDecoderListener> listener;
    std::shared_ptr<MessageDecoder> decoder;
  };
  auto state = std::make_shared<State>();
  state->listener = std::make_shared<AssignMessageDecoderListener>(&state->result);
  state->decoder = std::make_shared<MessageDecoder>(state->listener);

  if (metadata_length < state->decoder->next_required_size()) {
    return Status::Invalid("metadata_length should be at least ",
                           state->decoder->next_required_size());
  }
  return file->ReadAsync(context, offset, metadata_length + body_length)
      .Then([=](std::shared_ptr<Buffer> metadata) -> Result<std::shared_ptr<Message>> {
        if (metadata->size() < metadata_length) {
          return Status::Invalid("Expected to read ", metadata_length,
                                 " metadata bytes but got ", metadata->size());
        }
        ARROW_RETURN_NOT_OK(
            state->decoder->Consume(SliceBuffer(metadata, 0, metadata_length)));
        switch (state->decoder->state()) {
          case MessageDecoder::State::INITIAL:
            return std::move(state->result);
          case MessageDecoder::State::METADATA_LENGTH:
            return Status::Invalid("metadata length is missing. File offset: ", offset,
                                   ", metadata length: ", metadata_length);
          case MessageDecoder::State::METADATA:
            return Status::Invalid("flatbuffer size ",
                                   state->decoder->next_required_size(),
                                   " invalid. File offset: ", offset,
                                   ", metadata length: ", metadata_length);
          case MessageDecoder::State::BODY: {
            auto body = SliceBuffer(metadata, metadata_length, body_length);
            if (body->size() < state->decoder->next_required_size()) {
              return Status::IOError("Expected to be able to read ",
                                     state->decoder->next_required_size(),
                                     " bytes for message body, got ", body->size());
            }
            RETURN_NOT_OK(state->decoder->Consume(body));
            return std::move(state->result);
          }
          case MessageDecoder::State::EOS:
            return Status::Invalid("Unexpected empty message in IPC file format");
          default:
            return Status::Invalid("Unexpected state: ", state->decoder->state());
        }
      });
}

Status AlignStream(io::InputStream* stream, int32_t alignment) {
  ARROW_ASSIGN_OR_RAISE(int64_t position, stream->Tell());