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