Method read

libminifi/src/core/ProcessSession.cpp:281–316

}

int ProcessSession::read(const std::shared_ptr<core::FlowFile> &flow, InputStreamCallback *callback) {
  try {
    std::shared_ptr<ResourceClaim> claim = nullptr;

    if (flow->getResourceClaim() == nullptr) {
      // No existed claim for read, we throw exception
      logger_->log_debug("For %s, no resource claim but size is %d", flow->getUUIDStr(), flow->getSize());
      if (flow->getSize() == 0) {
        return 0;
      }
      throw Exception(FILE_OPERATION_EXCEPTION, "No Content Claim existed for read");
    }

    claim = flow->getResourceClaim();

    std::shared_ptr<io::BaseStream> stream = content_session_->read(claim);

    if (nullptr == stream) {
      throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile content for read");
    }

    stream->seek(flow->getOffset());

    auto ret = callback->process(stream);
    if (ret < 0) {
      throw Exception(FILE_OPERATION_EXCEPTION, "Failed to process flowfile content");
    }
    return gsl::narrow<int>(ret);
  } catch (std::exception &exception) {
    logger_->log_debug("Caught Exception %s", exception.what());
    throw;
  } catch (...) {
    logger_->log_debug("Caught Exception during process session read");
    throw;
  }
}

void ProcessSession::importFrom(io::InputStream&& stream, const std::shared_ptr<core::FlowFile> &flow) {
  importFrom(stream, flow);