event_pump.h

/*   
 *   (c) Copyright 2008 Philipp Skadorov (philipp_s@users.sourceforge.net)
 *
 *   This file is part of FREESECS.
 *
 *   FREESECS is free software: you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation, either version 3 of the License, or
 *   (at your option) any later version.
 *
 *   FREESECS 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 General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with FREESECS, see COPYING.
 *   If not, see <http://www.gnu.org/licenses/>.
 */
#ifndef _EVENT_PUMP_H
#define _EVENT_PUMP_H
#include <semaphore.h>
#include "tsqueue.h"
#include "log.h"

#define EVENT_PUMP_LOG_LEVEL 50

namespace freesecs
{
    template <class T>
    class threaded_event_pump_t
    {
    public:
        threaded_event_pump_t():_thread_id(0), _stop(0)
        {
            if(0 != sem_init(&_sem, 0, 0))
            {
                throw;
            }
            if(0 != pthread_create(&_thread_id, NULL, &threaded_event_pump_t::thread_func, (void*)this))
            {
                throw;
            }
        }
        virtual ~threaded_event_pump_t()
        {
            _stop = true;
            sem_post(&_sem);
            pthread_join(_thread_id, NULL);
            sem_destroy(&_sem);
        }
        threaded_event_pump_t<T>& operator << (const T& t)
        {
            _q.push(t);
            sem_post(&_sem);
            return *this;
        }

    protected:
        virtual int process(const T&) = 0;

    private:
        static void* thread_func(void *pdata)
        {
            threaded_event_pump_t<T> *pump = (threaded_event_pump_t<T>*)pdata;
            TRACE_DEBUG(EVENT_PUMP_LOG_LEVEL, 
                    "event pump: start loop, "
                    "this=0x%x, thread_id=%d", 
                    (int)pump, (int)pump->_thread_id);

            while(1)
            {
                sem_wait(&pump->_sem);
                if(pump->_stop)
                {
                    TRACE_DEBUG(EVENT_PUMP_LOG_LEVEL, "event pump: exit from thread");
                    pthread_exit(NULL);
                }
                if(true == pump->_q.empty())
                {
                    TRACE_DEBUG(EVENT_PUMP_LOG_LEVEL, "event pump: q is empty");
                    continue;
                }
                
                pump->process(pump->_q.front());
                pump->_q.pop();
            }
            return NULL;
        }
    private:
        pthread_t _thread_id;
        thread_safe_queue<T> _q;
        sem_t _sem;
        bool _stop;
    };


    template <class T>
    class simple_event_pump_t
    {
    public:
        simple_event_pump_t()
        {
            pthread_mutexattr_t attr;
            pthread_mutexattr_init(&attr);
            pthread_mutex_init(&_mutex, NULL);
        }

        virtual ~simple_event_pump_t()
        {
            pthread_mutex_destroy(&_mutex);
        }

        simple_event_pump_t<T>& operator << (const T& t)
        {
            lock_t l(_mutex);
            process(t);
            return *this;
        }

        int process_event(const T& t)
        {
            lock_t l(_mutex);
            return process(t);
        }

    protected:
        virtual int process(const T&) = 0;

    private:
        pthread_mutex_t _mutex;
    };


    #if 0
    //#define event_pump_t threaded_event_pump_t
    #define event_pump_t simple_event_pump_t
    #endif//if 0

}//namespace
#endif //_EVENT_PUMP_H

---