tcpMsgSocket.h

Go to the documentation of this file.

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


#ifndef tcpMsgSocket_H_
#define tcpMsgSocket_H_

#include "msgSocket.h"
#include "msgBundler.h"
#include "msgCompressor.h"
#include <list>
#include <vlutil/vlInetTcpSocket.h>

using namespace MAKRti;

class DtQueuedMsg
{
public:
   DtQueuedMsg() : msg(0), size(0), remainingMsg(0), remainingSize(0) {}
   DtQueuedMsg(caddr_t m, size_t s) : msg(m), size(s),
      remainingMsg(m), remainingSize(s) {}
   ~DtQueuedMsg() {}

   caddr_t msg;
   size_t size;
   caddr_t remainingMsg;
   size_t remainingSize;
};

class DT_DLL_RTIUTIL DtTcpMsgSocket : public DtMsgSocket
{
public:

   DtTcpMsgSocket(DtBoost::shared_ptr<DtMsgHeaderReader> hdrReader,
      size_t recvBufferSize = 20000, int maxQueuedMsgs = -1);

   DtTcpMsgSocket(std::auto_ptr<DtInetTcpSocket> socket,
      DtBoost::shared_ptr<DtMsgHeaderReader> hdrReader,
      size_t recvBuffSize = 20000, int maxQueuedMsgs = -1);

   virtual ~DtTcpMsgSocket();

   virtual void openServerSocket( DtU16 localPort,
      DtU32 flags = DtDefaultSockOpts, DtInetDevice* hostIf = NULL, 
      DtInetUtils::DtInetAddrFamily family = DtInetUtils::DtInetAddrFamily_IPv4 );

   virtual void openClientSocket( const DtInetEndpoint& dest,
      DtU32 flags = DtDefaultSockOpts, DtU16 localPort = 0,
      DtInetDevice* hostIf = NULL);

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

   virtual bool flush();

   virtual int recvMessage(caddr_t *buffptr);

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

   virtual void enableBundling(size_t maxBundleSize);

   virtual void disableBundling();

   virtual unsigned bundleSize() const;

   virtual void enableCompression(DtU8 compressionLevel);

   virtual void disableCompression();

   virtual unsigned compressionLevel() const;

   // Test the connection
   virtual bool testConnectionStatus();

   virtual void setMaxQueuedMessages(int size);
   
   virtual int maxQueuedMessages() const;

   virtual bool getQueueStatus(int& percentFull) const;

   virtual void setTcpNoDelay(bool enable);

   virtual void setNonBlocking(bool nonBlocking);

   virtual const DtInetEndpoint& destination() const;

   virtual DtU32 maxMsgSize() const;

   DtInetTcpSocket* inetTcpSocket();

   bool isConnected() const { return mySocket.get() && mySocket->state() == DtInetSockState_CONNECTED; }

   bool isConnectionInProgress() const { return mySocket.get() && mySocket->state() == DtInetSockState_INPROGRESS; }

protected:

   virtual int sendData(caddr_t msg, size_t msgSize, bool deleteAfterSend);

   virtual bool flushQueue();

   virtual bool queueMessage(caddr_t msg, size_t msgSize);
   
   virtual void resizeTestPacket();

protected:

   bool myNeedSize;

   int myNumBytesToWaitFor;

   int myNumBytesReceived;

   DtInetBuffer myTestPacket;

   std::list<DtQueuedMsg> myMsgQueue;

   DtMsgBundler myMsgBundler;

   bool myBundlingEnabled;

   DtMsgCompressor myMsgCompressor;

   bool myCompressionEnabled;

   int myMaxQueuedMessages;

   mutable int myMutableOldQueueLevel;

   size_t myBytesNeededForSize;
};

inline DtInetTcpSocket* DtTcpMsgSocket::inetTcpSocket()
{
   return static_cast<DtInetTcpSocket*>(mySocket.get());
}

#endif //! tcpMsgSocket_H_