| 602 | } |
| 603 | |
| 604 | Status WriteMessage(const Buffer& message, const IpcWriteOptions& options, |
| 605 | io::OutputStream* file, int32_t* message_length) { |
| 606 | const int32_t prefix_size = options.write_legacy_ipc_format ? 4 : 8; |
| 607 | const int32_t flatbuffer_size = static_cast<int32_t>(message.size()); |
| 608 | |
| 609 | int32_t padded_message_length = static_cast<int32_t>( |
| 610 | PaddedLength(flatbuffer_size + prefix_size, options.alignment)); |
| 611 | |
| 612 | int32_t padding = padded_message_length - flatbuffer_size - prefix_size; |
| 613 | |
| 614 | // The returned message size includes the length prefix, the flatbuffer, |
| 615 | // plus padding |
| 616 | *message_length = padded_message_length; |
| 617 | |
| 618 | // ARROW-6314: Write continuation / padding token |
| 619 | if (!options.write_legacy_ipc_format) { |
| 620 | RETURN_NOT_OK(file->Write(&internal::kIpcContinuationToken, sizeof(int32_t))); |
| 621 | } |
| 622 | |
| 623 | // Write the flatbuffer size prefix including padding in little endian |
| 624 | int32_t padded_flatbuffer_size = |
| 625 | bit_util::ToLittleEndian(padded_message_length - prefix_size); |
| 626 | RETURN_NOT_OK(file->Write(&padded_flatbuffer_size, sizeof(int32_t))); |
| 627 | |
| 628 | // Write the flatbuffer |
| 629 | RETURN_NOT_OK(file->Write(message.data(), flatbuffer_size)); |
| 630 | if (padding > 0) { |
| 631 | RETURN_NOT_OK(file->Write(kPaddingBytes, padding)); |
| 632 | } |
| 633 | |
| 634 | return Status::OK(); |
| 635 | } |
| 636 | |
| 637 | // ---------------------------------------------------------------------- |
| 638 | // Implement MessageDecoder |