Method get

libminifi/src/core/ProcessSession.cpp:912–955

}

std::shared_ptr<core::FlowFile> ProcessSession::get() {
  std::shared_ptr<Connectable> first = process_context_->getProcessorNode()->pickIncomingConnection();

  if (first == nullptr) {
    logger_->log_trace("Get is null for %s", process_context_->getProcessorNode()->getName());
    return nullptr;
  }

  std::shared_ptr<Connection> current = std::static_pointer_cast<Connection>(first);

  do {
    std::set<std::shared_ptr<core::FlowFile> > expired;
    std::shared_ptr<core::FlowFile> ret = current->poll(expired);
    if (!expired.empty()) {
      // Remove expired flow record
      for (const auto& record : expired) {
        std::stringstream details;
        details << process_context_->getProcessorNode()->getName() << " expire flow record " << record->getUUIDStr();
        provenance_report_->expire(record, details.str());
        // there is no rolling back expired FlowFiles
        if (record->isStored() && process_context_->getFlowFileRepository()->Delete(record->getUUIDStr())) {
          record->setStoredToRepository(false);
        }
      }
    }
    if (ret) {
      // add the flow record to the current process session update map
      ret->setDeleted(false);
      std::shared_ptr<FlowFile> snapshot = std::make_shared<FlowFileRecord>();
      *snapshot = *ret;
      logger_->log_debug("Create Snapshot FlowFile with UUID %s", snapshot->getUUIDStr());
      utils::Identifier uuid = ret->getUUID();
      _updatedFlowFiles[uuid] = {ret, snapshot};
      auto flow_version = process_context_->getProcessorNode()->getFlowIdentifier();
      if (flow_version != nullptr) {
        ret->setAttribute(SpecialFlowAttribute::FLOW_ID, flow_version->getFlowId());
      }
      return ret;
    }
    current = std::static_pointer_cast<Connection>(process_context_->getProcessorNode()->pickIncomingConnection());
  } while (current != nullptr && current != first);

  return nullptr;
}

void ProcessSession::flushContent() {
  content_session_->commit();