vlAsyncSocket.h

Go to the documentation of this file.

/*******************************************************************************
** Copyright (c) 2005 MaK Technologies, Inc.
** All rights reserved.
*******************************************************************************/
/*******************************************************************************
** $RCSfile: vlAsyncSocket.h,v $ $Revision: 1.2.6.1 $ $State: Exp $
*******************************************************************************/
#ifndef vlAsyncSocket_H_
#define vlAsyncSocket_H_


#ifndef VXWORKS

#include "vlNetSocket.h"
#include <vlutil/vlMutex.h>
#include <vlutil/vlThreadSafe.h>
#include <vlutil/vlThreadedObject.h>
#include <vlutil/vlRunnable.h>
#include <deque>



#ifdef DtUSE_UTILITIES_NAMESPACE
namespace DtUSE_UTILITIES_NAMESPACE
{
#endif

class DT_DLL_NETCOMPAT DtAsyncSocketPacket
{
public:
   DtAsyncSocketPacket();
   DtAsyncSocketPacket(int size, const char *data, DtIpv4Addr addr=0); 
   DtAsyncSocketPacket(const DtAsyncSocketPacket &other); 
   virtual ~DtAsyncSocketPacket();


   DtAsyncSocketPacket &operator=(const DtAsyncSocketPacket &other);

   size_t size() const { return mySize;}
   char * data()  { return &myData[0];}
   DtIpv4Addr addr() const { return myAddr;}

protected:

   int mySize;
   std::vector<char> myData;
   DtIpv4Addr myAddr;

};

class DT_DLL_NETCOMPAT DtAsyncNetSocket : public DtNetSocket
{
   public:

      DtAsyncNetSocket(int port = DtDEFAULT_PORT, 
                       DtU8 flags = DtReadWrite, DtIpv4Addr devAddr = 0);

      DtAsyncNetSocket(int port, 
                       DtIpv4Addr destAddr, DtU8 flags, 
                       DtIpv4Addr devAddr = 0);
    
      DtAsyncNetSocket(char* dummy, int sockfd, DtU8 flags = DtReadWrite);

      virtual ~DtAsyncNetSocket();

   private:

      DtAsyncNetSocket(const DtAsyncNetSocket &);

      DtAsyncNetSocket & operator=(const DtAsyncNetSocket &);

   public:

      virtual int sendTo(caddr_t packet, size_t size, DtIpv4Addr addr);

      virtual int directSendTo(caddr_t packet, size_t size, DtIpv4Addr addr);
      
      virtual bool usingAsyncSend() const;
      virtual void setUsingAsyncSend(bool b);
      
      virtual bool usingAsyncReceive() const;
      virtual void setUsingAsyncReceive(bool b);
      
   protected:

      virtual int recvFromNet(caddr_t *buffptr);

      void initThread();
      
      virtual void workerSendThread();
      
      virtual void workerReceiveThread();

      virtual void processSendQueue();
         
      virtual void processReceiveQueue();

   protected:
#ifdef _WIN32
      static unsigned long runReceive(void* usr_data);
      static unsigned long runSend(void *user_data);
#else
      static void* runReceive(void* usr_data);
      static void* runSend(void* usr_data);
#endif
      static void setQueueTimeout(double new_time_out);
      static double queueTimeout();

   protected:

      static  double theQueueTimeout;


   protected:
      
      typedef DtThreadSafe<std::deque<DtAsyncSocketPacket> > ThreadSafePacketQueue;

      bool myUseAsyncSendFlag;
      bool myUseAsyncReceiveFlag;

      ThreadSafePacketQueue mySendQueue;

      ThreadSafePacketQueue myReceiveQueue;

      volatile bool myTimeToQuitSender;
      volatile bool myTimeToQuitReceiver;
      char  *myThreadPktBuff;

#ifdef _WIN32
      HANDLE mySendThreadHandle;
      HANDLE myReceiveThreadHandle;
      HANDLE mySendEvent;
      HANDLE myReceiveEvent;
#else
      pthread_t mySendThreadHandle;
      pthread_t myReceiveThreadHandle;
      int mySendEvent[2];
#endif

};

inline  void 
DtAsyncNetSocket::setQueueTimeout(double time_out)
{
   theQueueTimeout = time_out;
}

inline double 
DtAsyncNetSocket::queueTimeout()
{
   return theQueueTimeout;
}
inline bool 
DtAsyncNetSocket::usingAsyncSend() const
{
   return myUseAsyncSendFlag;
}

inline void 
DtAsyncNetSocket::setUsingAsyncSend(bool b)
{
   myUseAsyncSendFlag = b;
}
inline bool 
DtAsyncNetSocket::usingAsyncReceive() const
{
   return myUseAsyncReceiveFlag;
}

inline void 
DtAsyncNetSocket::setUsingAsyncReceive(bool b)
{
   myUseAsyncReceiveFlag = b;
}

#ifdef DtUSE_UTILITIES_NAMESPACE
} 
#endif

#endif   //! !VXWORKS
#endif   //! DtAsyncNetSocket_H_