| 534 | } |
| 535 | |
| 536 | Status DecodeMessage(MessageDecoder* decoder, io::InputStream* file) { |
| 537 | if (decoder->state() == MessageDecoder::State::INITIAL) { |
| 538 | uint8_t continuation[sizeof(int32_t)]; |
| 539 | ARROW_ASSIGN_OR_RAISE(int64_t bytes_read, file->Read(sizeof(int32_t), &continuation)); |
| 540 | if (bytes_read == 0) { |
| 541 | // EOS without indication |
| 542 | return Status::OK(); |
| 543 | } else if (bytes_read != decoder->next_required_size()) { |
| 544 | return Status::Invalid("Corrupted message, only ", bytes_read, " bytes available"); |
| 545 | } |
| 546 | ARROW_RETURN_NOT_OK(decoder->Consume(continuation, bytes_read)); |
| 547 | } |
| 548 | |
| 549 | if (decoder->state() == MessageDecoder::State::METADATA_LENGTH) { |
| 550 | // Valid IPC message, read the message length now |
| 551 | uint8_t metadata_length[sizeof(int32_t)]; |
| 552 | ARROW_ASSIGN_OR_RAISE(int64_t bytes_read, |
| 553 | file->Read(sizeof(int32_t), &metadata_length)); |
| 554 | if (bytes_read != decoder->next_required_size()) { |
| 555 | return Status::Invalid("Corrupted metadata length, only ", bytes_read, |
| 556 | " bytes available"); |
| 557 | } |
| 558 | ARROW_RETURN_NOT_OK(decoder->Consume(metadata_length, bytes_read)); |
| 559 | } |
| 560 | |
| 561 | if (decoder->state() == MessageDecoder::State::EOS) { |
| 562 | return Status::OK(); |
| 563 | } |
| 564 | |
| 565 | auto metadata_length = decoder->next_required_size(); |
| 566 | ARROW_ASSIGN_OR_RAISE(auto metadata, file->Read(metadata_length)); |
| 567 | if (metadata->size() != metadata_length) { |
| 568 | return Status::Invalid("Expected to read ", metadata_length, " metadata bytes, but ", |
| 569 | "only read ", metadata->size()); |
| 570 | } |
| 571 | ARROW_RETURN_NOT_OK(decoder->Consume(metadata)); |
| 572 | |
| 573 | if (decoder->state() == MessageDecoder::State::BODY) { |
| 574 | ARROW_ASSIGN_OR_RAISE(auto body, file->Read(decoder->next_required_size())); |
| 575 | if (body->size() < decoder->next_required_size()) { |
| 576 | return Status::IOError("Expected to be able to read ", |
| 577 | decoder->next_required_size(), |
| 578 | " bytes for message body, got ", body->size()); |
| 579 | } |
| 580 | ARROW_RETURN_NOT_OK(decoder->Consume(body)); |
| 581 | } |
| 582 | |
| 583 | if (decoder->state() == MessageDecoder::State::INITIAL || |
| 584 | decoder->state() == MessageDecoder::State::EOS) { |
| 585 | return Status::OK(); |
| 586 | } else { |
| 587 | return Status::Invalid("Failed to decode message"); |
| 588 | } |
| 589 | } |
| 590 | |
| 591 | Result<std::unique_ptr<Message>> ReadMessage(io::InputStream* file, MemoryPool* pool) { |
| 592 | std::unique_ptr<Message> message; |