| 369 | } |
| 370 | |
| 371 | void ProcessSession::import(std::string source, const std::shared_ptr<FlowFile> &flow, bool keepSource, uint64_t offset) { |
| 372 | std::shared_ptr<ResourceClaim> claim = content_session_->create(); |
| 373 | size_t size = getpagesize(); |
| 374 | std::vector<uint8_t> charBuffer(size); |
| 375 | |
| 376 | try { |
| 377 | auto startTime = utils::timeutils::getTimeMillis(); |
| 378 | std::ifstream input; |
| 379 | input.open(source.c_str(), std::fstream::in | std::fstream::binary); |
| 380 | std::shared_ptr<io::BaseStream> stream = content_session_->write(claim); |
| 381 | |
| 382 | if (nullptr == stream) { |
| 383 | throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open new flowfile content for write"); |
| 384 | } |
| 385 | if (input.is_open() && input.good()) { |
| 386 | bool invalidWrite = false; |
| 387 | // Open the source file and stream to the flow file |
| 388 | if (offset != 0) { |
| 389 | input.seekg(offset); |
| 390 | if (!input.good()) { |
| 391 | logger_->log_error("Seeking to %d failed for file %s (does file/filesystem support seeking?)", offset, source); |
| 392 | invalidWrite = true; |
| 393 | } |
| 394 | } |
| 395 | while (input.good()) { |
| 396 | input.read(reinterpret_cast<char*>(charBuffer.data()), size); |
| 397 | if (input) { |
| 398 | if (stream->write(charBuffer.data(), gsl::narrow<int>(size)) < 0) { |
| 399 | invalidWrite = true; |
| 400 | break; |
| 401 | } |
| 402 | } else { |
| 403 | if (stream->write(reinterpret_cast<uint8_t*>(charBuffer.data()), gsl::narrow<int>(input.gcount())) < 0) { |
| 404 | invalidWrite = true; |
| 405 | break; |
| 406 | } |
| 407 | } |
| 408 | } |
| 409 | |
| 410 | if (!invalidWrite) { |
| 411 | flow->setSize(stream->size()); |
| 412 | flow->setOffset(0); |
| 413 | flow->setResourceClaim(claim); |
| 414 | |
| 415 | logger_->log_debug("Import offset %" PRIu64 " length %" PRIu64 " into content %s for FlowFile UUID %s", flow->getOffset(), flow->getSize(), flow->getResourceClaim()->getContentFullPath(), |
| 416 | flow->getUUIDStr()); |
| 417 | |
| 418 | stream->close(); |
| 419 | input.close(); |
| 420 | if (!keepSource) |
| 421 | std::remove(source.c_str()); |
| 422 | std::stringstream details; |
| 423 | details << process_context_->getProcessorNode()->getName() << " modify flow record content " << flow->getUUIDStr(); |
| 424 | auto endTime = utils::timeutils::getTimeMillis(); |
| 425 | provenance_report_->modifyContent(flow, details.str(), endTime - startTime); |
| 426 | } else { |
| 427 | stream->close(); |
| 428 | input.close(); |
no test coverage detected