Miroirs/xwords
1
0
Fork 0
mirror of git://xwords.git.sourceforge.net/gitroot/xwords/xwords synced 2025-01-03 23:04:08 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

move callback into queueelem so same queue can service udp and tcp (eventually)

Browse source
This commit is contained in:
Eric House 2013-01-25 07:19:44 -08:00
parent 4417591b06
commit 7ee402d2ae
2 changed files with 19 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
xwords4/relay/udpqueue.cpp
View file

@ -37,7 +37,6 @@ UdpThreadClosure::logStats()
}
UdpQueue::UdpQueue()
:m_cb(NULL)
{
pthread_mutex_init ( &m_queueMutex, NULL );
pthread_cond_init( &m_queueCondVar, NULL );
@ -68,9 +67,8 @@ void
UdpQueue::handle( const AddrInfo::AddrUnion* saddr, unsigned char* buf, int len,
QueueCallback cb )
{
UdpThreadClosure* utc = new UdpThreadClosure( saddr, buf, len );
UdpThreadClosure* utc = new UdpThreadClosure( saddr, buf, len, cb );
MutexLock ml( &m_queueMutex );
setCB( cb );
m_queue.push_back( utc );
pthread_cond_signal( &m_queueCondVar );
}
@ -88,7 +86,7 @@ UdpQueue::thread_main()
pthread_mutex_unlock( &m_queueMutex );
utc->noteDequeued();
(*m_cb)( utc );
(*utc->cb())( utc );
utc->logStats();
delete utc;
}
@ -104,9 +102,3 @@ UdpQueue::thread_main_static( void* closure )
return me->thread_main();
}
void
UdpQueue::setCB( QueueCallback cb )
{
assert( cb == m_cb || !m_cb );
m_cb = cb;
}

28
xwords4/relay/udpqueue.h
View file

@ -29,15 +29,23 @@
using namespace std;
class UdpThreadClosure;
typedef void (*QueueCallback)( UdpThreadClosure* closure );
class UdpThreadClosure {
public:
UdpThreadClosure( const AddrInfo::AddrUnion* saddr, unsigned char* buf, int len ) {
m_saddr = *saddr;
m_buf = new unsigned char[len];
memcpy( m_buf, buf, len );
m_len = len;
m_created = time( NULL );
}
UdpThreadClosure( const AddrInfo::AddrUnion* saddr, unsigned char* buf,
int len, QueueCallback cb )
: m_buf(new unsigned char[len])
, m_len(len)
, m_saddr(*saddr)
, m_cb(cb)
, m_created(time( NULL ))
{
memcpy( m_buf, buf, len );
}
~UdpThreadClosure() { delete m_buf; }
const unsigned char* buf() const { return m_buf; }
@ -45,17 +53,17 @@ public:
const AddrInfo::AddrUnion* saddr() const { return &m_saddr; }
void noteDequeued() { m_dequed = time( NULL ); }
void logStats();
const QueueCallback cb() const { return m_cb; }
private:
unsigned char* m_buf;
int m_len;
AddrInfo::AddrUnion m_saddr;
QueueCallback m_cb;
time_t m_created;
time_t m_dequed;
};
typedef void (*QueueCallback)( UdpThreadClosure* closure );
class UdpQueue {
public:
static UdpQueue* get();
@ -67,13 +75,11 @@ class UdpQueue {
private:
static void* thread_main_static( void* closure );
void* thread_main();
void setCB( QueueCallback cb );
pthread_mutex_t m_queueMutex;
pthread_cond_t m_queueCondVar;
deque<UdpThreadClosure*> m_queue;
QueueCallback m_cb;
};
#endif