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 "internalTypes.h"
#include <vlutil/vlInetSocket.h>
#include <vlutil/vlPrint.h>
#include <list>

#include "msgHeaderReader.h"

class DT_DLL_RTIUTIL DtMsgSocket
{
public:

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

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

   virtual ~DtMsgSocket();

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

   virtual bool flush() = 0;

   virtual int recvMessage(caddr_t *buffptr) = 0;

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

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

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

   const MAKRti::DtString& lastError() const;

   virtual void enableBundling(size_t maxBundleSize) = 0;

   virtual void disableBundling() = 0;

   virtual void enableCompression(MAKRti::DtU8 compressionLevel) = 0;

   virtual void disableCompression() = 0;

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

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

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

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

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

   MAKRti::DtInetSocket* inetSocket();

private:

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

protected:

   void expandRecvBuffer( size_t sizeNeeded );

protected:

   DtUniquePtr<MAKRti::DtInetSocket> mySocket;

   MAKRti::DtInetBuffer myRecvBuffer;

   MAKRti::DtString myLastError;

   std::shared_ptr<DtMsgHeaderReader> myHeaderReader;

};

inline MAKRti::DtInetSockState DtMsgSocket::state() const
{
   MAKRti::DtInetSockState state = MAKRti::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 MAKRti::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 MAKRti::DtInetSocket* DtMsgSocket::inetSocket()
{
   return mySocket.get();
}

#endif