Delimit IPC stream messages and reassemble with the indicated messages included. This way we can remove messages from an IPC stream to test different failure modes or other difficult-to-test behaviors
| 2179 | // included. This way we can remove messages from an IPC stream to test |
| 2180 | // different failure modes or other difficult-to-test behaviors |
| 2181 | void SpliceMessages(std::shared_ptr<Buffer> stream, |
| 2182 | const std::vector<int>& included_indices, |
| 2183 | std::shared_ptr<Buffer>* spliced_stream) { |
| 2184 | ASSERT_OK_AND_ASSIGN(auto out, io::BufferOutputStream::Create(0)); |
| 2185 | |
| 2186 | io::BufferReader buffer_reader(stream); |
| 2187 | std::unique_ptr<MessageReader> message_reader = MessageReader::Open(&buffer_reader); |
| 2188 | std::unique_ptr<Message> msg; |
| 2189 | |
| 2190 | // Parse and reassemble first two messages in stream |
| 2191 | int message_index = 0; |
| 2192 | while (true) { |
| 2193 | ASSERT_OK_AND_ASSIGN(msg, message_reader->ReadNextMessage()); |
| 2194 | if (!msg) { |
| 2195 | break; |
| 2196 | } |
| 2197 | |
| 2198 | if (std::find(included_indices.begin(), included_indices.end(), message_index++) == |
| 2199 | included_indices.end()) { |
| 2200 | // Message being dropped, continue |
| 2201 | continue; |
| 2202 | } |
| 2203 | |
| 2204 | IpcWriteOptions options; |
| 2205 | IpcPayload payload; |
| 2206 | payload.type = msg->type(); |
| 2207 | payload.metadata = msg->metadata(); |
| 2208 | payload.body_buffers.push_back(msg->body()); |
| 2209 | payload.body_length = msg->body()->size(); |
| 2210 | int32_t unused_metadata_length = -1; |
| 2211 | ASSERT_OK(ipc::WriteIpcPayload(payload, options, out.get(), &unused_metadata_length)); |
| 2212 | } |
| 2213 | ASSERT_OK_AND_ASSIGN(*spliced_stream, out->Finish()); |
| 2214 | } |
| 2215 | |
| 2216 | TEST(TestRecordBatchStreamReader, NotEnoughDictionaries) { |
| 2217 | // ARROW-6126 |
no test coverage detected