tcpMsgSocketMgr.h

Go to the documentation of this file.

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


#ifndef tcpMsgSocketMgr_H_
#define tcpMsgSocketMgr_H_

#include <set>
#include <vlutil/vlInetSocketMgr.h>
#include "tcpMsgSocket.h"

typedef void (*DtTcpSocketAdditionCallback)(DtTcpMsgSocket* sock, void* usr);
typedef DtTcpSocketAdditionCallback DtTcpSocketRemovalCallback;

class DtTcpSocketMgrCbInfo
{
public:

   DtTcpSocketMgrCbInfo(DtTcpSocketAdditionCallback cb, void* usr)
   {
      myCb = cb;
      myUsr = usr;
   }

   bool operator==(const DtTcpSocketMgrCbInfo& rhs)
   {
      return (myCb == rhs.myCb && myUsr == rhs.myUsr);
   }

   DtTcpSocketAdditionCallback myCb;
   void* myUsr;
};

typedef std::map<MAKRti::DtInetTcpSocket*, DtTcpMsgSocket*> DtTcpMsgSocketMap;

class DT_DLL_RTIUTIL DtTcpMsgSocketMgr
{
public:

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

   virtual ~DtTcpMsgSocketMgr();

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

   void processNewConnections();

   virtual void processConnections();

   virtual void addSocket(DtTcpMsgSocket* sock);

   static bool socketAdded(MAKRti::DtInetSocket* sock, void* usr);
   bool socketAdded(MAKRti::DtInetSocket* sock);

   DtTcpMsgSocketMap::iterator checkIfReconnecting(const MAKRti::DtInetTcpSocket * newSock);
   virtual void addSocketAdditionCallback(DtTcpSocketAdditionCallback cb, void* usr);
   virtual void removeSocketAdditionCallback(DtTcpSocketAdditionCallback cb,  void* usr);

   static bool socketRemoved(MAKRti::DtInetSocket* sock, void* usr);
   bool socketRemoved(MAKRti::DtInetSocket* sock);

   virtual void addSocketRemovalCallback(DtTcpSocketRemovalCallback cb, void* usr);
   virtual void removeSocketRemovalCallback(DtTcpSocketRemovalCallback cb, void* usr);

   virtual void removeAndDelete(DtTcpMsgSocket* sock);

   virtual void enableBundling(size_t maxBundleSize);

   virtual void disableBundling();

   virtual void enableCompression(MAKRti::DtU8 compressionLevel);

   virtual void disableCompression();

   virtual void setTcpNoDelay(bool enable);

   virtual void setSendBufferSize(MAKRti::DtU32 size);

   virtual void setRecvBufferSize(MAKRti::DtU32 size);

   virtual MAKRti::DtInetSockState serverSocketState() const;

   virtual MAKRti::DtU32 maxMsgSize() const;

   DtTcpMsgSocketMap& socketSet();

   DtTcpMsgSocketMap::iterator socketIter(DtTcpMsgSocket* sock);

   MAKRti::DtString getLastError() const;

   bool isListening() const { return serverSocketState() == MAKRti::DtInetSockState_LISTENING; }

private:

   DtTcpMsgSocketMgr(const DtTcpMsgSocketMgr&);
   DtTcpMsgSocketMgr& operator=(const DtTcpMsgSocketMgr&);

protected:

   DtUniquePtr<MAKRti::DtInetSocketMgr> mySocketMgr;

   DtTcpMsgSocketMap mySockets;

   std::shared_ptr<DtMsgHeaderReader> myHeaderReader;

   int myMaxQueuedMessages;

   size_t myRecvBufferSize;

   bool myTcpNoDelay;

   size_t myMaxBundleSize;

   MAKRti::DtU8 myCompressionLevel;

   MAKRti::DtU32 mySockRecvBufferSize;

   MAKRti::DtU32 mySockSendBufferSize;

   std::list<DtTcpSocketMgrCbInfo> myAdditionCallbacks;

   std::list<DtTcpSocketMgrCbInfo> myRemovalCallbacks;

   MAKRti::DtString myLastError;

   static const MAKRti::DtU32 theDefaultSockOpts;
   static const MAKRti::DtU32 theSockOptNonBlocking;
};

inline DtTcpMsgSocketMap& DtTcpMsgSocketMgr::socketSet()
{
   return mySockets;
}

inline DtTcpMsgSocketMap::iterator DtTcpMsgSocketMgr::socketIter(DtTcpMsgSocket* sock)
{
   DtTcpMsgSocketMap::iterator iter = mySockets.end();
   if(sock)
   {
      MAKRti::DtInetTcpSocket* inetSock = sock->inetTcpSocket();
      if(inetSock)
      {
         iter = mySockets.find(inetSock);
      }
   }
   return iter;
}

inline MAKRti::DtString DtTcpMsgSocketMgr::getLastError() const
{
   return myLastError;
}

#endif