Method ConsumeBuffer

cpp/src/arrow/ipc/message.cc:719–763

  }

  Status ConsumeBuffer(std::shared_ptr<Buffer> buffer) {
    if (buffered_size_ == 0) {
      while (buffer->size() >= next_required_size_) {
        auto used_size = next_required_size_;
        switch (state_) {
          case State::INITIAL:
            RETURN_NOT_OK(ConsumeInitialBuffer(buffer));
            break;
          case State::METADATA_LENGTH:
            RETURN_NOT_OK(ConsumeMetadataLengthBuffer(buffer));
            break;
          case State::METADATA:
            if (buffer->size() == next_required_size_) {
              return ConsumeMetadataBuffer(buffer);
            } else {
              auto sliced_buffer = SliceBuffer(buffer, 0, next_required_size_);
              RETURN_NOT_OK(ConsumeMetadataBuffer(sliced_buffer));
            }
            break;
          case State::BODY:
            if (buffer->size() == next_required_size_) {
              return ConsumeBodyBuffer(buffer);
            } else {
              auto sliced_buffer = SliceBuffer(buffer, 0, next_required_size_);
              RETURN_NOT_OK(ConsumeBodyBuffer(sliced_buffer));
            }
            break;
          case State::EOS:
            return Status::OK();
        }
        if (buffer->size() == used_size) {
          return Status::OK();
        }
        buffer = SliceBuffer(buffer, used_size);
      }
    }

    if (buffer->size() == 0) {
      return Status::OK();
    }

    buffered_size_ += buffer->size();
    chunks_.push_back(std::move(buffer));
    return ConsumeChunks();
  }

  int64_t next_required_size() const { return next_required_size_ - buffered_size_; }