Method startProcessingProcessors

libminifi/src/core/ProcessGroup.cpp:141–191

}

void ProcessGroup::startProcessingProcessors(const std::shared_ptr<TimerDrivenSchedulingAgent>& timeScheduler,
    const std::shared_ptr<EventDrivenSchedulingAgent> &eventScheduler, const std::shared_ptr<CronDrivenSchedulingAgent> &cronScheduler) {
  std::unique_lock<std::recursive_mutex> lock(mutex_);

  std::set<std::shared_ptr<Processor> > failed_processors;

  for (const auto &processor : failed_processors_) {
    try {
      logger_->log_debug("Starting %s", processor->getName());
      switch (processor->getSchedulingStrategy()) {
        case TIMER_DRIVEN:
          timeScheduler->schedule(processor);
          break;
        case EVENT_DRIVEN:
          eventScheduler->schedule(processor);
          break;
        case CRON_DRIVEN:
          cronScheduler->schedule(processor);
          break;
      }
    }
    catch (const std::exception &e) {
      logger_->log_error("Failed to start processor %s (%s): %s", processor->getUUIDStr(), processor->getName(), e.what());
      failed_processors.insert(processor);
    }
    catch (...) {
      logger_->log_error("Failed to start processor %s (%s)", processor->getUUIDStr(), processor->getName());
      failed_processors.insert(processor);
    }
  }
  failed_processors_ = std::move(failed_processors);

  for (auto& processor : failed_processors_) {
    try {
      processor->onUnSchedule();
    } catch (...) {
      logger_->log_error("Exception occured during unscheduling processor: %s (%s)", processor->getUUIDStr(), processor->getName());
    }
  }

  if (!onScheduleTimer_ && !failed_processors_.empty() && onschedule_retry_msec_ > 0) {
    logger_->log_info("Retrying failed processors in %lld msec", onschedule_retry_msec_.load());
    auto func = [this, eventScheduler, cronScheduler, timeScheduler]() {
      this->startProcessingProcessors(timeScheduler, eventScheduler, cronScheduler);
    };
    onScheduleTimer_.reset(new utils::CallBackTimer(std::chrono::milliseconds(onschedule_retry_msec_), func));
    onScheduleTimer_->start();
  } else if (failed_processors_.empty() && onScheduleTimer_) {
    onScheduleTimer_->stop();
  }
}

void ProcessGroup::startProcessing(const std::shared_ptr<TimerDrivenSchedulingAgent>& timeScheduler, const std::shared_ptr<EventDrivenSchedulingAgent> &eventScheduler,
                                   const std::shared_ptr<CronDrivenSchedulingAgent> &cronScheduler) {