Method onTrigger

extensions/standard-processors/processors/GetFile.cpp:159–203

}

void GetFile::onTrigger(core::ProcessContext* /*context*/, core::ProcessSession *session) {
  // Perform directory list

  metrics_->iterations_++;

  const bool isDirEmptyBeforePoll = isListingEmpty();
  logger_->log_debug("Is listing empty before polling directory %i", isDirEmptyBeforePoll);
  if (isDirEmptyBeforePoll) {
    if (request_.pollInterval == 0 || (utils::timeutils::getTimeMillis() - last_listing_time_) > request_.pollInterval) {
      performListing(request_);
      last_listing_time_.store(utils::timeutils::getTimeMillis());
    }
  }

  const bool isDirEmptyAfterPoll = isListingEmpty();
  logger_->log_debug("Is listing empty after polling directory %i", isDirEmptyAfterPoll);

  if (!isDirEmptyAfterPoll) {
    try {
      std::queue<std::string> list;
      pollListing(list, request_);
      while (!list.empty()) {
        std::string fileName = list.front();
        list.pop();
        logger_->log_info("GetFile process %s", fileName);
        auto flowFile = session->create();
        if (flowFile == nullptr)
          return;
        std::size_t found = fileName.find_last_of("/\\");
        std::string path = fileName.substr(0, found);
        std::string name = fileName.substr(found + 1);
        flowFile->setAttribute(core::SpecialFlowAttribute::FILENAME, name);
        flowFile->setAttribute(core::SpecialFlowAttribute::PATH, path);
        flowFile->addAttribute(core::SpecialFlowAttribute::ABSOLUTE_PATH, fileName);
        session->import(fileName, flowFile, request_.keepSourceFile);
        session->transfer(flowFile, Success);
      }
    } catch (std::exception &exception) {
      logger_->log_debug("GetFile Caught Exception %s", exception.what());
      throw;
    } catch (...) {
      throw;
    }
  }
}

bool GetFile::isListingEmpty() {
  std::lock_guard<std::mutex> lock(mutex_);