pipemsgqueue.h

Generic template class library for message queue implementation, uses pthread mutex.

#ifndef __PIPEMSGQUEUE_H__
#define __PIPEMSGQUEUE_H__

#include <TMQ_Pipe.h>
#include <Message.h>
#include <unistd.h>
#include <sys/time.h>

#define WRITE_PIPE_DESC 1
#define READ_PIPE_DESC 0

template <class T>
class PipeContainer
{
    public:
	PipeContainer() ;
	virtual ~PipeContainer() ;

	void post(T& theData) ;
	T get(struct timeval* ptrTimeVal = NULL) ;

    private:
	int filedesc[2] ;
	int max_filedesc;
};

class PipeMsgQueue : public TMQ_Pipe<Message*, PipeContainer<Message*> >
{
} ;


template <class T>
PipeContainer<T>::PipeContainer()
: max_filedesc(0) 
{
    pipe(filedesc) ;

    if (
      ( filedesc[ WRITE_PIPE_DESC ] )
      > max_filedesc
      )
    {
      max_filedesc = filedesc[ WRITE_PIPE_DESC ];
    }

    if (
      ( filedesc[ READ_PIPE_DESC ] )
      > max_filedesc
      )
    {
      max_filedesc = filedesc[ READ_PIPE_DESC ];
    }

}

template <class T>
PipeContainer<T>::~PipeContainer() 
{
    close( filedesc[ WRITE_PIPE_DESC ] );
    close( filedesc[ READ_PIPE_DESC ] );
}

template <class T>
void PipeContainer<T>::post(T& theData )
{
    size_t bytesWritten = write(
	filedesc[ WRITE_PIPE_DESC ],
	&theData,
	sizeof(theData) );
}

template <class T>
T PipeContainer<T>::get(struct timeval* ptrTimeVal )
{
    T returnValue = (T)0;
    fd_set read_fdset;
    int numberOfDescriptors = max_filedesc + 1;

    FD_ZERO( &read_fdset );
    FD_SET( filedesc[ READ_PIPE_DESC ], &read_fdset );

    int selectResult = select(
                            numberOfDescriptors,
                            &read_fdset,
                            NULL,
                            NULL,
			    ptrTimeVal
                            );

    if( selectResult == -1 )
    {
	// Interrupted, normally by SIGTERM signal and which normally means stop servicepath
	return (T)0 ;
    }

    if ( FD_ISSET( filedesc[ READ_PIPE_DESC ], &read_fdset  ) )
    {
	ssize_t bytesRead = read(
                            filedesc[ READ_PIPE_DESC ],
                            &returnValue,
                            sizeof( returnValue ) ) ;

    }

    return returnValue ;
}

#endif /* __PIPEMSGQUEUE_H__ */