Method DoExchange

cpp/src/arrow/flight/test_definitions.cc:1260–1279

  }

  Status DoExchange(const ServerCallContext& context,
                    std::unique_ptr<FlightMessageReader> reader,
                    std::unique_ptr<FlightMessageWriter> writer) override {
    FlightStreamChunk chunk;
    bool begun = false;
    while (true) {
      ARROW_ASSIGN_OR_RAISE(chunk, reader->Next());
      if (!chunk.data) break;
      if (!begun) {
        begun = true;
        RETURN_NOT_OK(writer->Begin(chunk.data->schema()));
      }
      RETURN_NOT_OK(CheckBuffersOnDevice(*chunk.data, *device_));
      // XXX: do not assume transport will synchronize, we must
      // synchronize or else data will be "missing"
      RETURN_NOT_OK(context_->Synchronize());
      RETURN_NOT_OK(writer->WriteRecordBatch(*chunk.data));
    }
    return Status::OK();
  }

  const RecordBatchVector& batches() const { return batches_; }