| 33 | namespace minifi { |
| 34 | |
| 35 | utils::TaskRescheduleInfo CronDrivenSchedulingAgent::run(const std::shared_ptr<core::Processor> &processor, const std::shared_ptr<core::ProcessContext> &processContext, |
| 36 | const std::shared_ptr<core::ProcessSessionFactory> &sessionFactory) { |
| 37 | if (this->running_ && processor->isRunning()) { |
| 38 | auto uuid = processor->getUUID(); |
| 39 | std::chrono::system_clock::time_point result; |
| 40 | std::chrono::system_clock::time_point from = std::chrono::system_clock::now(); |
| 41 | { |
| 42 | std::lock_guard<std::mutex> locK(mutex_); |
| 43 | |
| 44 | auto sched_f = schedules_.find(uuid); |
| 45 | if (sched_f != std::end(schedules_)) { |
| 46 | result = last_exec_[uuid]; |
| 47 | if (from >= result) { |
| 48 | result = sched_f->second.cron_to_next(from); |
| 49 | last_exec_[uuid] = result; |
| 50 | } else { |
| 51 | // we may be woken up a little early so that we can honor our time. |
| 52 | // in this case we can return the next time to run with the expectation |
| 53 | // that the wakeup mechanism gets more granular. |
| 54 | return utils::TaskRescheduleInfo::RetryIn(std::chrono::duration_cast<std::chrono::milliseconds>(result - from)); |
| 55 | } |
| 56 | } else { |
| 57 | Bosma::Cron schedule(processor->getCronPeriod()); |
| 58 | result = schedule.cron_to_next(from); |
| 59 | last_exec_[uuid] = result; |
| 60 | schedules_.insert(std::make_pair(uuid, schedule)); |
| 61 | } |
| 62 | } |
| 63 | |
| 64 | if (result > from) { |
| 65 | bool shouldYield = this->onTrigger(processor, processContext, sessionFactory); |
| 66 | |
| 67 | if (processor->isYield()) { |
| 68 | // Honor the yield |
| 69 | return utils::TaskRescheduleInfo::RetryIn(std::chrono::milliseconds(processor->getYieldTime())); |
| 70 | } else if (shouldYield && this->bored_yield_duration_ > 0) { |
| 71 | // No work to do or need to apply back pressure |
| 72 | return utils::TaskRescheduleInfo::RetryIn(std::chrono::milliseconds(this->bored_yield_duration_)); |
| 73 | } |
| 74 | } |
| 75 | return utils::TaskRescheduleInfo::RetryIn(std::chrono::duration_cast<std::chrono::milliseconds>(result - from)); |
| 76 | } |
| 77 | return utils::TaskRescheduleInfo::Done(); |
| 78 | } |
| 79 | |
| 80 | } /* namespace minifi */ |
| 81 | } /* namespace nifi */ |
nothing calls this directly
no test coverage detected