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>

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(std::shared_ptr<DtMsgHeaderReader> hdrReader,
      size_t recvBufferSize = 20000, int maxQueuedMsgs = -1);

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

   virtual ~DtTcpMsgSocket();

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

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

   virtual MAKRti::DtInetTcpSocket* reconnect();
   virtual void replaceSocket(DtUniquePtr<MAKRti::DtInetTcpSocket> newSock);

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

   virtual bool flush();

   virtual int recvMessage(caddr_t *buffptr);

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

   virtual void enableBundling(size_t maxBundleSize);

   virtual void disableBundling();

   virtual unsigned bundleSize() const;

   virtual void enableCompression(MAKRti::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 MAKRti::DtInetEndpoint& destination() const;

   virtual MAKRti::DtU32 maxMsgSize() const;

   MAKRti::DtInetTcpSocket* inetTcpSocket();

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

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

   MAKRti::DtString getErrorString() const;

   virtual bool isClient() const;

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;

   size_t myNumBytesToWaitFor;

   int myNumBytesReceived;

   MAKRti::DtInetBuffer myTestPacket;

   std::list<DtQueuedMsg> myMsgQueue;

   DtMsgBundler myMsgBundler;

   bool myBundlingEnabled;

   DtMsgCompressor myMsgCompressor;

   bool myCompressionEnabled;

   int myMaxQueuedMessages;

   mutable int myMutableOldQueueLevel;

   size_t myBytesNeededForSize;

   bool myIsClient;

   // These are client connection related params required for reconnection
   MAKRti::DtInetEndpoint myClientDestination;
   MAKRti::DtU32 myClientFlags;
   MAKRti::DtU16 myClientLocalPort;
   DtUniquePtr<MAKRti::DtInetDevice> myClientHostIf;

   static const MAKRti::DtU32 theDefaultSockOpts;
};

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

#endif