| 487 | } |
| 488 | |
| 489 | Future<std::shared_ptr<Message>> ReadMessageAsync(int64_t offset, int32_t metadata_length, |
| 490 | int64_t body_length, |
| 491 | io::RandomAccessFile* file, |
| 492 | const io::IOContext& context) { |
| 493 | // Make a std::shared_ptr so as to have a stable ReadMessageState pointer, |
| 494 | // since state->listener will point back to state->result. |
| 495 | auto state = std::make_shared<ReadMessageState>(); |
| 496 | |
| 497 | if (metadata_length < state->decoder->next_required_size()) { |
| 498 | return Status::Invalid("metadata_length should be at least ", |
| 499 | state->decoder->next_required_size()); |
| 500 | } |
| 501 | return file |
| 502 | ->ReadAsync(context, offset, metadata_length + body_length, |
| 503 | /*allow_short_read=*/false) |
| 504 | .Then([=](std::shared_ptr<Buffer> metadata) -> Result<std::shared_ptr<Message>> { |
| 505 | // Pass a nullptr file to ensure that no further IO occurs |
| 506 | // (we have fetched all the required bytes). |
| 507 | return ReadMessageContinued(offset, metadata_length, body_length, metadata, |
| 508 | /*file=*/nullptr, |
| 509 | /*fields_loader=*/{}, state.get()); |
| 510 | }); |
| 511 | } |
| 512 | |
| 513 | Status AlignStream(io::InputStream* stream, int32_t alignment) { |
| 514 | ARROW_ASSIGN_OR_RAISE(int64_t position, stream->Tell()); |