/* * * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information * regarding copyright ownership. The ASF licenses this file * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, * software distributed under the License is distributed on an * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * KIND, either express or implied. See the License for the * specific language governing permissions and limitations * under the License. * */ #include "Timer.h" #include using qpid::sys::AbsTime; using qpid::sys::Duration; using qpid::sys::Monitor; using qpid::sys::Thread; using namespace qpid::broker; TimerTask::TimerTask(Duration timeout) : duration(timeout), time(AbsTime::now(), timeout), cancelled(false) {} TimerTask::TimerTask(AbsTime _time) : duration(0), time(_time), cancelled(false) {} TimerTask::~TimerTask(){} void TimerTask::reset() { time.reset(AbsTime::now(), duration); } Timer::Timer() : active(false) { start(); } Timer::~Timer() { stop(); } void Timer::run() { Monitor::ScopedLock l(monitor); while(active){ if (tasks.empty()) { monitor.wait(); } else { TimerTask::shared_ptr t = tasks.top(); if (t->cancelled) { tasks.pop(); } else if(t->time < AbsTime::now()) { tasks.pop(); t->fire(); } else { monitor.wait(t->time); } } } } void Timer::add(TimerTask::shared_ptr task) { Monitor::ScopedLock l(monitor); tasks.push(task); monitor.notify(); } void Timer::start() { Monitor::ScopedLock l(monitor); if (!active) { active = true; runner = Thread(this); } } void Timer::stop() { signalStop(); runner.join(); } void Timer::signalStop() { Monitor::ScopedLock l(monitor); if (active) { active = false; monitor.notifyAll(); } } bool Later::operator()(const TimerTask::shared_ptr& a, const TimerTask::shared_ptr& b) const { return a.get() && b.get() && a->time > b->time; } bool LaterA::operator()(const TimerTaskA::intrusive_ptr& a, const TimerTaskA::intrusive_ptr& b) const { return a.get() && b.get() && a->time > b->time; } TimerA::TimerA(): active(false) { start(); } TimerA::~TimerA() { stop(); } void TimerA::run() { Monitor::ScopedLock l(monitor); while(active){ if (itasks.empty()) { monitor.wait(); } else { TimerTaskA::intrusive_ptr t = itasks.top(); if (t->cancelled) { itasks.pop(); } else if(t->time < AbsTime::now()) { itasks.pop(); t->fire(); } else { monitor.wait(t->time); } } } // ::run(); } TimerTaskA::TimerTaskA(qpid::sys::Duration timeout): TimerTask(timeout), ref_cnt(0) {} TimerTaskA::TimerTaskA(qpid::sys::AbsTime time): TimerTask(time), ref_cnt(0) {} TimerTaskA::~TimerTaskA() {} void TimerA::add(TimerTaskA::intrusive_ptr& task) { Monitor::ScopedLock l(monitor); itasks.push(task); monitor.notify(); } void TimerA::start() { Monitor::ScopedLock l(monitor); if (!active) { active = true; runner = Thread(this); } } void TimerA::stop() { signalStop(); runner.join(); } void TimerA::signalStop() { Monitor::ScopedLock l(monitor); if (active) { active = false; monitor.notifyAll(); } } void qpid::broker::intrusive_ptr_add_ref(TimerTaskA* fe) { fe->ref(); } void qpid::broker::intrusive_ptr_release(TimerTaskA* fe) { fe->unref(); if (fe->refcnt() == 0) delete fe; }