| 661 | skip_body_(skip_body) {} |
| 662 | |
| 663 | Status ConsumeData(const uint8_t* data, int64_t size) { |
| 664 | if (buffered_size_ == 0) { |
| 665 | while (size > 0 && size >= next_required_size_) { |
| 666 | auto used_size = next_required_size_; |
| 667 | switch (state_) { |
| 668 | case State::INITIAL: |
| 669 | RETURN_NOT_OK(ConsumeInitialData(data, next_required_size_)); |
| 670 | break; |
| 671 | case State::METADATA_LENGTH: |
| 672 | RETURN_NOT_OK(ConsumeMetadataLengthData(data, next_required_size_)); |
| 673 | break; |
| 674 | case State::METADATA: { |
| 675 | // We need to copy metadata because it's used in |
| 676 | // ConsumeBody(). ConsumeBody() may be called from another |
| 677 | // ConsumeData(). We can't assume that the given data for |
| 678 | // the current ConsumeData() call is still valid in the |
| 679 | // next ConsumeData() call. So we need to copy metadata |
| 680 | // here. |
| 681 | ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Buffer> buffer, |
| 682 | AllocateBuffer(next_required_size_, pool_)); |
| 683 | memcpy(buffer->mutable_data(), data, next_required_size_); |
| 684 | RETURN_NOT_OK(ConsumeMetadataBuffer(buffer)); |
| 685 | } break; |
| 686 | case State::BODY: { |
| 687 | // We don't need to copy the given data for body because |
| 688 | // we can assume that a decoded record batch should be |
| 689 | // valid only in a listener_->OnMessageDecoded() call. If |
| 690 | // the passed message is needed to be valid after the |
| 691 | // call, it's a listener_'s responsibility. The listener_ |
| 692 | // may copy the data for it. |
| 693 | auto buffer = std::make_shared<Buffer>(data, next_required_size_); |
| 694 | RETURN_NOT_OK(ConsumeBodyBuffer(buffer)); |
| 695 | } break; |
| 696 | case State::EOS: |
| 697 | return Status::OK(); |
| 698 | } |
| 699 | data += used_size; |
| 700 | size -= used_size; |
| 701 | } |
| 702 | } |
| 703 | |
| 704 | if (size == 0) { |
| 705 | return Status::OK(); |
| 706 | } |
| 707 | |
| 708 | // We need to copy unused data because the given data for the |
| 709 | // current ConsumeData() call may be invalid in the next |
| 710 | // ConsumeData() call. |
| 711 | ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Buffer> chunk, AllocateBuffer(size, pool_)); |
| 712 | memcpy(chunk->mutable_data(), data, size); |
| 713 | chunks_.push_back(std::move(chunk)); |
| 714 | buffered_size_ += size; |
| 715 | return ConsumeChunks(); |
| 716 | } |
| 717 | |
| 718 | Status ConsumeBuffer(std::shared_ptr<Buffer> buffer) { |
| 719 | if (buffered_size_ == 0) { |
no test coverage detected