Method onTrigger

extensions/libarchive/BinFiles.cpp:259–331

}

void BinFiles::onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) {
  // Rollback is not viable for this processor!!
  {
    // process resurrected FlowFiles first
    auto flowFiles = file_store_.getNewFlowFiles();
    // these are already processed FlowFiles, that we own
    bool hadFailure = false;
    for (auto &file : flowFiles) {
      std::string groupId = getGroupId(context.get(), file);
      bool offer = this->binManager_.offer(groupId, file);
      if (!offer) {
        session->transfer(file, Failure);
        hadFailure = true;
      } else {
        // no need to route successfully captured such files as we already own them
      }
    }
    if (hadFailure) {
      context->yield();
      return;
    }
  }

  for (size_t i = 0; i < batchSize_; ++i) {
    auto flow = session->get();

    if (flow == nullptr) {
      break;
    }

    preprocessFlowFile(context.get(), session.get(), flow);
    std::string groupId = getGroupId(context.get(), flow);

    bool offer = this->binManager_.offer(groupId, flow);
    if (!offer) {
      session->transfer(flow, Failure);
      context->yield();
      return;
    }
    // assuming ownership over the incoming flowFile
    session->transfer(flow, Self);
  }

  // migrate bin to ready bin
  this->binManager_.gatherReadyBins();
  if (this->binManager_.getBinCount() > maxBinCount_) {
    // bin count reach max allowed
    context->yield();
    logger_->log_debug("BinFiles reach max bin count %d", this->binManager_.getBinCount());
    this->binManager_.removeOldestBin();
  }

  // get the ready bin
  std::deque<std::unique_ptr<Bin>> readyBins;
  binManager_.getReadyBin(readyBins);

  // process the ready bin
  while (!readyBins.empty()) {