//--------------------------msgdequeue.h开始-------------------------------------
//实现可控队列
#ifndef MSGDEQUEUE_H
#define MSGDEQUEUE_H
#include <errno.h>
#include <time.h>
#include <semaphore.h>
#include <deque>
#include <iostream>
#include "tmutex.h"
using namespace std;
typedef ThreadMutex MUTEX_TYPE;
template<typename T, typename MUTEX_TYPE>
class CMessageDequeue
{
public:
CMessageDequeue(size_t MaxSize) : m_MaxSize( MaxSize )
{
sem_init( &m_enques,0, m_MaxSize ); //入队信号量初始化为MaxSize，最多可容纳MaxSize各元素
sem_init( &m_deques,0,0 ); //队列刚开始为空，出队信号量初始为0
}
~CMessageDequeue()
{
sem_destroy(&m_enques);
sem_destroy(&m_deques);
}
int sem_wait_i( sem_t *psem, int mswait )
{//等待信号量变成>0，mswait为等待时间，若mswait<0则无穷等待，否则等待若干mswait毫秒。
if( mswait < 0 ){
int rv = 0;
while( ((rv = sem_wait(psem) ) != 0 ) && (errno == EINTR) );
//等待信号量，errno==EINTR屏蔽其他信号事件引起的等待中断
return rv;
}
else {
timespec ts;
clock_gettime(CLOCK_REALTIME, &ts ); //获取当前时间
ts.tv_sec += (mswait / 1000 ); //加上等待时间的秒数
ts.tv_nsec += ( mswait % 1000 ) * 1000; //加上等待时间纳秒数
int rv = 0;
while( ((rv=sem_timedwait( psem, &ts ))!=0) && (errno ==EINTR) );
//等待信号量，errno==EINTR屏蔽其他信号事件引起的等待中断
return rv;
}
}
bool push_back( const T &item, int mswait = -1 )
{ //等待mswait毫秒直到将item插入队列，mswait为-1则一直等待
if( -1 == sem_wait_i( &m_enques, mswait )) {
return false;
}
//AUTO_GUARD：定界加锁，见tmutex.h文件定义。
//AUTO_GUARD( g, MUTEX_TYPE, m_lock );
CAutoGuard< MUTEX_TYPE > g(m_lock);
try{
m_data.push_back( item );
//cout << "push " << item << endl;
sem_post( &m_deques );
return true;
}
catch(const exception& e){
cout<<e.what()<<endl;
return false;
}
}
bool pop_front( T &item, bool bpop = true, int mswait = -1 ) {
//等待mswait毫秒直到从队列取出元素，mswait为-1则一直等待
if( -1 == sem_wait_i( &m_deques, mswait ) ) {
return false;
}
//#define AUTO_GUARD( guard_tmp_var, MUTEX_TYPE, mtx)CAutoGuard<(MUTEX_TYPE)> guard_tmp_var((mtx))
//AUTO_GUARD：定界加锁，见tmutex.h文件定义。
//AUTO_GUARD( g, MUTEX_TYPE, m_lock ) ;
CAutoGuard< MUTEX_TYPE > g(m_lock);
try
{
item = m_data.front();
if( bpop ){
m_data.pop_front();
// cout << "pop " << item << endl;
}
sem_post( &m_enques );
return true;
}
catch(const exception& e)
{
cout <<e.what()<<endl;
return false;
}
}
inline size_t size()
{
return m_data.size();
}
private:
MUTEX_TYPE m_lock;
deque<T> m_data;
size_t m_MaxSize;
sem_t m_enques;
sem_t m_deques;
};
#endif
