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