Method TestDoExchangeConcurrency

cpp/src/arrow/flight/test_definitions.cc:575–604

  // would immediately return an error.
}
void DataTest::TestDoExchangeConcurrency() {
  // Ensure that we can do reads/writes on separate threads
  auto descr = FlightDescriptor::Command("echo");
  ASSERT_OK_AND_ASSIGN(auto exchange, client_->DoExchange(descr));
  std::unique_ptr<FlightStreamReader> reader = std::move(exchange.reader);
  std::unique_ptr<FlightStreamWriter> writer = std::move(exchange.writer);

  RecordBatchVector batches;
  ASSERT_OK(ExampleIntBatches(&batches));
  ASSERT_OK(writer->Begin(ExampleIntSchema()));

  std::thread reader_thread([&reader, &batches]() {
    for (size_t i = 0; i < batches.size(); i++) {
      ASSERT_OK_AND_ASSIGN(auto chunk, reader->Next());
      ASSERT_NE(nullptr, chunk.data);
      ASSERT_EQ(nullptr, chunk.app_metadata);
      AssertBatchesEqual(*batches[i], *chunk.data);
    }
    ASSERT_OK_AND_ASSIGN(auto chunk, reader->Next());
    ASSERT_EQ(nullptr, chunk.data);
    ASSERT_EQ(nullptr, chunk.app_metadata);
  });

  for (const auto& batch : batches) {
    ASSERT_OK(writer->WriteRecordBatch(*batch));
  }
  ASSERT_OK(writer->DoneWriting());
  reader_thread.join();
  ASSERT_OK(writer->Close());
}
void DataTest::TestDoExchangeUndrained() {
  // Ensure if the application doesn't drain all messages, that the
  // server/transport does