| 419 | } |
| 420 | |
| 421 | Status Begin(const std::shared_ptr<Schema>& schema, |
| 422 | const ipc::IpcWriteOptions& options) override { |
| 423 | if (batch_writer_) { |
| 424 | return Status::Invalid("This writer has already been started."); |
| 425 | } |
| 426 | std::unique_ptr<ipc::internal::IpcPayloadWriter> payload_writer( |
| 427 | new ClientPutPayloadWriter(stream_, std::move(descriptor_), |
| 428 | write_size_limit_bytes_, &app_metadata_)); |
| 429 | // XXX: this does not actually write the message to the stream. |
| 430 | // See Close(). |
| 431 | |
| 432 | // On failure, we should close the stream to make sure we get any gRPC-side error |
| 433 | auto status = |
| 434 | ipc::internal::OpenRecordBatchWriter(std::move(payload_writer), schema, options) |
| 435 | .Value(&batch_writer_); |
| 436 | if (!status.ok()) { |
| 437 | closed_ = true; |
| 438 | final_status_ = stream_->Finish(std::move(status)); |
| 439 | return final_status_; |
| 440 | } |
| 441 | return Status::OK(); |
| 442 | } |
| 443 | |
| 444 | Status Begin(const std::shared_ptr<Schema>& schema) override { |
| 445 | return Begin(schema, write_options_); |