msgSocket.h

Go to the documentation of this file.

/*********************************************************************
** Copyright (c) 2010 MaK Technologies, Inc.
** All rights reserved.
*********************************************************************/
/*********************************************************************
** $RCSfile: msgSocket.h,v $ $Revision: $ $State: Exp $
*********************************************************************/


#ifndef msgSocket_H_
#define msgSocket_H_

#include <rtiMsConfig.h>
#include <vlutil/vlInetSocket.h>
#include <vlutil/vlPrint.h>
#include <list>

#include "msgHeaderReader.h"

using namespace MAKRti;

class DT_DLL_RTIUTIL DtMsgSocket
{
public:

   DtMsgSocket(DtBoost::shared_ptr<DtMsgHeaderReader> hdrReader,
      size_t recvBuffSize);

   DtMsgSocket(std::auto_ptr<DtInetSocket> socket,
      DtBoost::shared_ptr<DtMsgHeaderReader> hdrReader, size_t recvBuffSize);

   virtual ~DtMsgSocket();

   virtual int sendMessage(caddr_t msg, size_t msgSize, bool deleteAfterSend,
      const DtInetAddr& destAddr = DtInetAddr::inaddrAny(), int port = 0) = 0;

   virtual bool flush() = 0;

   virtual int recvMessage(caddr_t *buffptr) = 0;

   virtual int recvMessageWithSrc(caddr_t *buffptr, DtInetEndpoint& src) = 0;

   virtual bool closeSocket();
      
   virtual DtInetSockState state() const;

   void printStateToStream(MAKRti::DtOutputStream& stream) const;

   const DtString& lastError() const;

   virtual void enableBundling(size_t maxBundleSize) = 0;

   virtual void disableBundling() = 0;

   virtual void enableCompression(DtU8 compressionLevel) = 0;

   virtual void disableCompression() = 0;

   const DtInetEndpoint& localEndpoint() const { return mySocket->localEndpoint(); }

   void setSendBufferSize(DtU32 size) { mySocket->setSendBufferSize(size); }

   void setRecvBufferSize(DtU32 size) { mySocket->setReceiveBufferSize(size); }

   DtU32 maxSendMsgSize() const { return mySocket->getSendBufferSize(); }

   unsigned int descriptor() const { return mySocket->descriptor(); }

   DtInetSocket* inetSocket();

private:

   DtMsgSocket(const DtMsgSocket&);
   DtMsgSocket& operator=(const DtMsgSocket&);

protected:

   void expandRecvBuffer( size_t sizeNeeded );

protected:

   std::auto_ptr<MAKRti::DtInetSocket> mySocket;

   DtInetBuffer myRecvBuffer;

   DtString myLastError;

   DtBoost::shared_ptr<DtMsgHeaderReader> myHeaderReader;

};

inline DtInetSockState DtMsgSocket::state() const
{
   DtInetSockState state = DtInetSockState_NOT_INITIALIZED;
   if(mySocket.get())
   {
      state = mySocket->state();
   }
   return state;   
}

inline void DtMsgSocket::printStateToStream(MAKRti::DtOutputStream& stream) const
{
   MAKRti::DtString stateStr = "Socket State = ";
   stateStr += state();
   stream << stateStr.c_str() << std::endl;
   if (!lastError().isEmpty())
   {
      stream << lastError().c_str() << std::endl;
   }
}

inline const DtString& DtMsgSocket::lastError() const
{
   return myLastError;
}

inline void DtMsgSocket::expandRecvBuffer( size_t sizeNeeded )
{
   if(sizeNeeded > myRecvBuffer.size())
   {
      div_t ratio = div( (int)sizeNeeded, (int)myRecvBuffer.size() );
      size_t newSize = myRecvBuffer.size() * ( ratio.quot + 1 );
      myRecvBuffer.resize(newSize);
   }
}

inline DtInetSocket* DtMsgSocket::inetSocket()
{
   return mySocket.get();
}

#endif