---

receiver.h

Go to the documentation of this file.

#ifndef UTILS_RECEIVER_H
#define UTILS_RECEIVER_H

#include <list>

#include <boost/thread/xtime.hpp>
#include <boost/thread/condition.hpp>
#include <boost/thread/thread.hpp>

#define warn nowarn             // ignore warnings

using namespace std;

namespace utils {


template <class T, class F, int TIMEOUT = 100>
class receiver: private boost::noncopyable
{
        F func;
        
        list<T> queue;

        boost::mutex            mutex;
    boost::condition    not_empty;

        bool                            please_quit;
        boost::mutex            please_quit_mutex;

        boost::thread           thread;
public:

        receiver();
        

        void shutdown();


        void add(const T& v);

private:        
        // proceed next items in the list (if any)
        void checkItems();
        
        friend class Runner;
        
        class Runner
        {
        public:
                receiver& rec;
        
                Runner(receiver& r): rec(r)
                        { }
                
                void operator()();
        };
};

template <class T, class F, int TIMEOUT>
receiver<T,F,TIMEOUT>::receiver(): please_quit(false), thread(Runner(*this))
        { }

template <class T, class F, int TIMEOUT>
void receiver<T,F,TIMEOUT>::shutdown()
{ 
        {
                warn("please quit.");
                boost::mutex::scoped_lock lock(please_quit_mutex);                      
                please_quit = true;
        }

        thread.join();
}


// adds an item to the receiver
template <class T, class F, int TIMEOUT>
void receiver<T,F,TIMEOUT>::add(const T& v)
{       
        warn("adding an item.");
        boost::mutex::scoped_lock lock(mutex);
        queue.push_back(v);

        not_empty.notify_one();
}

// proceed next items in the list (if any)
template <class T, class F, int TIMEOUT>
void receiver<T,F,TIMEOUT>::checkItems()
{
        warn("checkitems");
        boost::mutex::scoped_lock lock(mutex);
        if(queue.size() == 0) 
        {
                warn("checkitems: waiting for not_empty");
            boost::xtime xt;
        boost::xtime_get(&xt, boost::TIME_UTC);
            xt.sec += TIMEOUT / 1000;           
        xt.nsec += (TIMEOUT % 1000) * 1000000;
                if(xt.nsec >= 1000000000) {
                        xt.sec += 1;
                        xt.nsec -= 1000000000;
                }

                not_empty.timed_wait(lock, xt);
        }

        int cnt = queue.size();
        int i = cnt;
        while(i--) {
                warn("run func");
                func(*queue.begin());
                warn("ok");
                queue.pop_front();
        }
        
        warn("checkitems done: %d items.", cnt);
}

template <class T, class F, int TIMEOUT>
void receiver<T,F,TIMEOUT>::Runner::operator()()
{
        while(1)
        {
                rec.checkItems();

                boost::mutex::scoped_lock lock(rec.please_quit_mutex);
                if(rec.please_quit)
                {
                        warn("Please quit? -> ok.");
                        break;
                }
        }

        rec.checkItems();
}

};  // namespace utils

#undef warn

#endif

---