TimeoutProvider.h

Go to the documentation of this file.

/*
  Copyright (C) 2006, 2005, 2004 Erik Eliasson, Johan Bilien, Werner Dittmann

  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation; either
  version 2.1 of the License, or (at your option) any later version.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  Lesser General Public License for more details.

  You should have received a copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/


#ifndef TIMEOUTPROVIDER_H
#define TIMEOUTPROVIDER_H

#include <list>
#include <sys/time.h>

#include <cc++/config.h>
#include <cc++/thread.h>

template <class TOCommand, class TOSubscriber>
class TPRequest
{

public:

    TPRequest( TOSubscriber tsi, int timeoutMs, const TOCommand &command):
        subscriber(tsi)
    {
        struct timeval tv;
        gettimeofday(&tv, NULL );

        when_ms = ((uint64)tv.tv_sec) * (uint64)1000 + ((uint64)tv.tv_usec) / (uint64)1000;
        when_ms += timeoutMs;
        this->command = command;
    }

    bool happensBefore(uint64 t)
    {
        if (when_ms < t) {
            return true;
        }
        if (when_ms > t) {
            return false;
        }
        return false; // if equal it does not "happens_before"

    }

    bool happensBefore(const TPRequest *req){
        return happensBefore(req->when_ms);
    }

    int getMsToTimeout ()
    {
        struct timeval tv;
        gettimeofday(&tv, NULL );

        uint64 now = ((uint64)tv.tv_sec) * (uint64)1000 + ((uint64)tv.tv_usec) / (uint64)1000;

        if (happensBefore(now)) {
            return 0;
        }
        else {
            return (int)(when_ms - now);
        }
    }

    TOCommand getCommand()
    {
        return command;
    }

    TOSubscriber getSubscriber()
    {
        return subscriber;
    }

    bool operator==(const TPRequest<TOCommand, TOSubscriber> &req)
    {
        if (req.subscriber == subscriber &&
               req.command == command &&
               req.when_ms == when_ms) {
            return true;
        }
        return false;
    }

private:
    TOSubscriber subscriber;
    uint64 when_ms;             // Time since Epoch in ms when the timeout
                                // will happen

    TOCommand command;          // Command that will be delivered to the
                                // receiver (subscriber) of the timeout.
};

template<class TOCommand, class TOSubscriber>
        class TimeoutProvider : public ost::Thread, ost::Event {

public:

    TimeoutProvider(): requests(), synchLock(), stop(false)  { }

    ~TimeoutProvider() {
        terminate();
    }

    void stopThread(){
        stop = true;
        signal();               // signal event to waiting thread
    }

    void requestTimeout(int32_t time_ms, TOSubscriber subscriber, const TOCommand &command)
    {
        TPRequest<TOCommand, TOSubscriber>* request =
                new TPRequest<TOCommand, TOSubscriber>(subscriber, time_ms, command);

        synchLock.enter();

        if (requests.size()==0) {
            requests.push_front(request);
            signal();
            synchLock.leave();
            return;
        }
        if (request->happensBefore(requests.front())) {
            requests.push_front(request);
            signal();
            synchLock.leave();
            return;
        }
        if (requests.back()->happensBefore(request)){
            requests.push_back(request);
            signal();
            synchLock.leave();
            return;
        }

        typename std::list<TPRequest<TOCommand, TOSubscriber>* >::iterator i;
        for(i = requests.begin(); i != requests.end(); i++ ) {
            if( request->happensBefore(*i)) {
                requests.insert(i, request);
                break;
            }
        }
        signal();
        synchLock.leave();
    }

    void cancelRequest(TOSubscriber subscriber, const TOCommand &command)
    {
        synchLock.enter();
        typename std::list<TPRequest<TOCommand, TOSubscriber>* >::iterator i;
        for(i = requests.begin(); i != requests.end(); ) {
            if( (*i)->getCommand() == command &&
                (*i)->getSubscriber() == subscriber) {
                i = requests.erase(i);
                continue;
            }
            i++;
        }
        synchLock.leave();
    }

protected:

    void run()
    {
        do {
            synchLock.enter();
            int32_t time = 3600000;
            int32_t size = 0;
            if ((size = requests.size()) > 0) {
                time = requests.front()->getMsToTimeout();
            }
            if (time == 0 && size > 0) {
                if (stop){      // This must be checked so that we will
                                // stop even if we have timeouts to deliver.
                    return;
                }
                TPRequest<TOCommand, TOSubscriber>* req = requests.front();
                TOSubscriber subs = req->getSubscriber();
                TOCommand command = req->getCommand();

                requests.pop_front();

                synchLock.leave(); // call the command with free Mutex
                subs->handleTimeout(command);
                continue;
            }
            synchLock.leave();
            if (stop) {         // If we were told to stop while delivering
                                // a timeout we will exit here
                return;
            }
            reset();            // ready to receive triggers again
            wait(time);
            if (stop) {         // If we are told to exit while waiting we
                                // will exit
                return;
            }
        } while(true);
    }

private:

    // The timeouts are ordered in the order of which they
    // will expire. Nearest in future is first in list.
    std::list<TPRequest<TOCommand, TOSubscriber> *> requests;

    ost::Mutex synchLock;       // Protects the internal data structures

    bool stop;          // Flag to tell the worker thread
                        // to terminate. Set to true and
                        // wake the worker thread to
                        // terminate it.
};

#endif

---

© sourcejam.com 2005-2008