MPNLBase.h

Go to the documentation of this file.

/***************************************************************************
                          mpnlbase.h  -  description
                             -------------------
    begin                : Sat Feb 7 2004
    copyright            : (C) 2004 by Mickael Faivre-Macon
    email                : mickael@easyplay.com.tw
 ***************************************************************************/

/***************************************************************************
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 ***************************************************************************/

#ifndef MPNLBASEH
#define MPNLBASEH

#include "MThread.h"
#include <exception>
#include <string>
#include <list>
#include <queue>
#include <sstream>
#include "MUtils.h"

#ifdef WIN32
   #include<winsock2.h>

   #define MYSOCKET SOCKET
   #define MYINVALID_SOCKET INVALID_SOCKET
   #define MYSOCKET_ERROR SOCKET_ERROR
   #define MYCLOSE(s) closesocket(s)
   #define SLEEP(x) Sleep(x)
#else
   #include <netinet/in.h>
   #include <arpa/inet.h> // inet_ntoa
   #include <sys/time.h>
   #include <netdb.h>

   #define MYSOCKET int
   #define MYINVALID_SOCKET -1
   #define MYSOCKET_ERROR -1
   #define MYCLOSE(s) close(s)
   #define SLEEP(x) usleep((x)*1000)
#endif

namespace MPNL {

//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
// Global Fonctions

std::string GetLocalIP();
std::string GetHostName(int socket);

//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
// Types
enum MMode {MODE_LENPREFIXED, MODE_LINE, MODE_PACKET};
//class error : public std::exception {};
class MPNLBase;
class MSocket;
struct MClient {
   MClient () : Socket(NULL), /*Thread(NULL),*/ ToBeDeleted(false) {}
   MSocket*       Socket;
   //boost::thread* Thread;
   bool ToBeDeleted;
   unsigned long ID;
   };
typedef std::list<MClient> MClientList;

class MTransferManager;

//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
//---------------------------------------------------------------------------

class MSocket : public MThread {

public:
   void*          Data;
   MMode          Mode;
   int            PrefixLen;
   struct sockaddr_in ServAddr; 
   std::string    PeerIP, LocalIP; 
   int            PeerPort;
   std::iostream*  Stream;

   MSocket(int type, MPNLBase* owner);
   virtual ~MSocket();

   void           operator()(); // thread function
   bool           IsInvalid();
   bool           Read(std::string& msg);
   bool           Read(std::string& ip, std::string& msg); // for UDP
   virtual void   Stop();
   int            Send(const std::string& msg);
   unsigned long  SendStream();
   void           SetStream(std::iostream* s) {Stream = s;}
   int            SendTo(const std::string& ip, int port, const std::string& msg);
   MYSOCKET       GetS() const {return S;}
   void           SetS(MYSOCKET s) {S=s;}
   bool           Listen(int port);
   //bool           IsActive();
   bool           HasMsg();
   std::string    GetLocalIP();

protected:
   typedef std::queue<std::string> MMsgQueue;

   //boost::mutex   Mutex;
   MYSOCKET       S;
   int            Type;
   MPNLBase*      Owner;
   //bool           Active;
   MMsgQueue      Messages;
   MMsgQueue      IP; // for UDP
   TIME           LastMsgTime;
   boost::mutex   MsgMutex;

   //   boost::mutex::scoped_lock lock(Mutex);
   //   Active = a;
   //   }
   void PushMsg(const std::string& msg);
   void PushMsg(const std::string& ip, const std::string& msg); // for UDP
   int Send(const char* str, int len, int flags);

   };

//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
class MPNLBase {

public:
   int         Port;

   MPNLBase();
   virtual ~MPNLBase();

   virtual void SetMode(MMode m)=0;
   virtual void SetPrefixLen(int len)=0;
   virtual void SetStream(std::iostream* s)=0;
   virtual bool OnConnection(MSocket* s)=0; 
   virtual void OnDisconnection(MSocket* s)=0;
   virtual void PushMsg(MSocket*) {};   // public because MSocket call this. Maybe change this.
   void SetLastError(const std::string& a) {
      boost::mutex::scoped_lock lock(Mutex);
      LastError = a;
      }
   std::string GetLastError() {
      boost::mutex::scoped_lock lock(Mutex);
      return LastError;
      }

protected:
   typedef std::deque<MSocket*> MMsgQueue;
   MMsgQueue      Messages;
   MMode          Mode;
   int            PrefixLen;

private:
   std::string    LastError;
   boost::mutex   Mutex;

   };

//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
class MTCPServer : public MPNLBase, public MThread /*accepting connections*/ {

public:
   struct MStats {
      MStats() {Total = CurCon = MaxCon = 0;}
      unsigned long Total, CurCon, MaxCon;
      };

   MTCPServer();
   virtual ~MTCPServer();

   void     operator()(); // thread function accepting connections
   bool     Listen();
   bool     Read(MSocket*& s);
   void     Disconnect(MSocket* s);
   virtual bool OnConnection(MSocket* s); // return false to reject the connection
   virtual void OnDisconnection(MSocket* s);
   virtual void SetMode(MMode m);
   virtual void SetPrefixLen(int len);
   virtual void SetStream(std::iostream* s);
   virtual void Stop();
/*
   void     SetActive(bool a) {
      boost::mutex::scoped_lock lock(Mutex);
      Active = a;
      }
   bool     IsActive() {
      boost::mutex::scoped_lock lock(Mutex);
      return Active;
      }
*/
   void     GetStats(MStats& s);
   void     BroadcastMsg(const std::string& msg);
   int      Send(const std::string& ip, const std::string& msg);
   bool     IsConnected(const std::string& ip);
   int      NbClients() { // or use getstats
      boost::mutex::scoped_lock lock(ClientMutex);
      return Clients.size();
      }
   MClientList* LockClients() {
      if(ClientLock->locked()) return NULL;
      ClientLock->lock();
      return &Clients;
      }
   void UnlockClients() {
      ClientLock->unlock();
      }
//*/

protected:
   boost::mutex         ClientMutex;
   MStats               Stats;
private:
   MClientList          Clients;
   boost::mutex::scoped_lock* ClientLock;
   boost::mutex         MsgMutex;
   MSocket*             Socket;
   //boost::thread*       Thread;
   //boost::thread_group  ClientThreads; // reading threads
   //bool                 Active;
   std::iostream*       Stream;

   virtual void         PushMsg(MSocket* s);
   bool                 SocketHasMsg(MSocket* s);

   };

//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
class MTCPClient : public MPNLBase {

public:
   std::string Host;

   MTCPClient();
   virtual ~MTCPClient();

   bool   Connect();
   void   Disconnect();


   int    Send(const std::string& msg);
   unsigned long  SendStream();

   bool   Read(std::string& msg);
   bool   OnConnection(MSocket* s);
   void   OnDisconnection(MSocket* s);
   bool   IsConnected() {return Socket->IsActive();}
   void   SetMode(MMode m) {Mode = m;Socket->Mode = m;}
   void   SetPrefixLen(int len);
   virtual void SetStream(std::iostream* s);
   std::string GetLocalIP() {return Socket->GetLocalIP();}

private:
   //boost::mutex         Mutex;
   MSocket*             Socket;
   //boost::thread*       Thread; // reading socket thread
   struct sockaddr_in LocalAddr;
   struct sockaddr_in ServAddr;


   };

//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
class MUDPServer : public MPNLBase, public MThread {

public:
   MUDPServer();
   virtual ~MUDPServer();

   void           operator()(); // thread function
   virtual bool   Listen(int port);
   virtual bool   Read(MSocket*& s);
   //virtual void   Stop();
   //virtual bool   OnConnection(MSocket* s); // return false to reject the connection
   //virtual void   OnDisconnection(MSocket* s);
   virtual void   SetMode(MMode m);
   //void           SetActive(bool a) {
   //   boost::mutex::scoped_lock lock(Mutex);
   //   Active = a;
   //   }
   //void           GetStats(MStats& s);

private:
   //boost::mutex         Mutex;
   boost::mutex         MsgMutex;
   MSocket*             Socket;
   //boost::thread*       Thread; // Socket thread
   bool                 Active;
   //MStats               Stats;

   void                 PushMsg(MSocket* s);
   bool                 SocketHasMsg();

   };

//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
class MUDPClient : public MPNLBase {

public:
   std::string Host;

   MUDPClient();
   virtual ~MUDPClient();

   bool PrepareToSend();
   
   virtual int    Send(const std::string& msg);

   virtual void   SetMode(MMode m) {Mode = m;Socket->Mode = m;}

private:
   MSocket*             Socket;

   };


//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
enum MTransferDirection {tdDOWNLOAD=0, tdUPLOAD=1};
enum MTransferMode {tmCLIENT, tmSERVER, tmBOTH};
class MTransfer;
typedef std::list<MTransfer*> MTransferList;

//---------------------------------------------------------------------------
class MTransferTCPServer : public MTCPServer {

public:
   MTransferDirection Direction;
   MTransfer* Transfer;

   MTransferTCPServer(MTransfer* t) : MTCPServer() {Transfer=t;}
   ~MTransferTCPServer() {}


   bool OnConnection(MSocket* s) {
      if(Direction==tdDOWNLOAD) return true;
      boost::mutex::scoped_lock lock(ClientMutex);
      Stats.Total++; // we depend on it
      lock.unlock();
      s->SendStream();  // TODO: should test the return of send stream with the len of the file
      SetActive(false);
      return false;
      }
   void OnDisconnection(MSocket* s);
   };

//---------------------------------------------------------------------------
class MTransferTCPClient : public MTCPClient {

public:
   MTransfer* Transfer;

   MTransferTCPClient(MTransfer* t) : MTCPClient() {Transfer=t;}
   ~MTransferTCPClient() {}

   void OnDisconnection(MSocket* s);

   };

//---------------------------------------------------------------------------
class MTransfer : public MThread {
   friend class MTransferManager;
public:
   class MUpdate {
      public:
         MUpdate() : Stamp(0) {}
         MUpdate(MTransfer* t) : Stamp(0), PeerIP(t->PeerIP), ID(t->ID) {}
/*
         MUpdate(const MUpdate& u) {
            boost::mutex::scoped_lock lock(u.Mutex);
            Type  = u.Type;
            Msg   = u.Msg;
            Stamp = u.Stamp;
            }
         MUpdate& operator=(const MUpdate& u) {
            if (this == &u) return *this;
            boost::mutex::scoped_lock lock1(&Mutex < &u.Mutex ? Mutex : u.Mutex);
            boost::mutex::scoped_lock lock2(&Mutex > &u.Mutex ? Mutex : u.Mutex);
            Type  = u.Type;
            Msg   = u.Msg;
            Stamp = u.Stamp;
            // Progress
            return *this;
            }
*/
         enum MType {utADD, utPROGRESS, utMSG, utREMOVE} Type;
         enum MResult {urOK, urFAILED} Result;
         std::string Msg, PeerIP;
         int ID;
         MTransferDirection Direction;
         unsigned char GetStamp() {
            boost::mutex::scoped_lock lock(Mutex);
            return Stamp;
            }
         void SetStamp(unsigned char s) {
            boost::mutex::scoped_lock lock(Mutex);
            Stamp |= s;
            }
         //unsigned long Progress;

      private:
         /*mutable*/ boost::mutex   Mutex;
         unsigned char Stamp; // when == Manager::ClearStamp, it is erased by the manager
      };
   typedef std::deque<MUpdate*> MUpdateList;

   MTransferMode  Mode;
   std::string    PeerIP;
   std::string    LocalPath;
   std::string    MD5;
   std::string    URL;
   int            Port;
   int            ID;
   bool           Dead;

   MTransfer();
   ~MTransfer();

   void Stop(void);  
   MTransferDirection GetDirection() {return Direction;}
   void SetDirection(MTransferDirection dir) {
      Direction = dir;
      Server->Direction = dir;
      }
   //bool IsActive() {
   //   boost::mutex::scoped_lock lock(Mutex);
   //   return Active;
   //   }
   bool IsPending() {
      boost::mutex::scoped_lock lock(AccessMutex);
      return Pending;
      }
   void SetPending(bool b) {
      boost::mutex::scoped_lock lock(AccessMutex);
      Pending = b;
      }
   /*const std::string& GetError() {
      boost::mutex::scoped_lock lock(Mutex);
      return Error;
      }*/
   void AddUpdate(MUpdate* u) {
      boost::mutex::scoped_lock lock(UpdateMutex);
      Updates.push_back(u);
      }

   MUpdateList* LockUpdates() {
      UpdateLock->lock();
      return &Updates;
      }
   void UnlockUpdates() {
      UpdateLock->unlock();
      }
      
   int NbUpdates() {
      boost::mutex::scoped_lock lock(UpdateMutex);
      return Updates.size();
      }

   std::string Mode2String() {
      switch(Mode) {
         case tmCLIENT: return "Client";
         case tmSERVER: return "Server";
         case tmBOTH: return "Both";
         default: return "Unknown";
         }
      }
   std::string Dir2String() {
      switch(Direction) {
         case tdUPLOAD: return "Upload";
         case tdDOWNLOAD: return "Download";
         default: return "Unknown";
         }
      }
   void Disconnected() {SetActive(false);}
   void EraseUpdate(MUpdateList::iterator& ite);

   void operator()(); 

private:
   //bool                 Active;
   bool                 Pending; 
   MTransferTCPServer*  Server;
   MTransferTCPClient*  Client;
   //boost::thread*       Thread;
   MTransferDirection   Direction;
   //std::string          Error;
   MUpdateList          Updates;
   boost::mutex::scoped_lock* UpdateLock;
   boost::mutex         UpdateMutex;
   boost::mutex         AccessMutex;

   //void Start(void); ///< Start negociations (thread). Moved to private recently.

   //void SetActive(bool b) {
   //   boost::mutex::scoped_lock lock(Mutex);
   //   Active = b;
   //   }
   /*void SetError(const std::string& s) {
      boost::mutex::scoped_lock lock(Mutex);
      Error = s;
      }*/

   };

inline  void MTransferTCPServer::OnDisconnection(MSocket* s) {
   Transfer->Disconnected();
   }
inline  void MTransferTCPClient::OnDisconnection(MSocket* s) {
   Transfer->Disconnected();
   }

//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
//---------------------------------------------------------------------------

class MTransferManager : public MThread {

public:

   struct MStats {
      MStats() : Total(0),Cur(0),Max(0) {}
      unsigned int Total, Cur, Max, DL, UP;
      } Stats;

   unsigned char ClearStamp;

   MTransferManager();
   ~MTransferManager();

   int StartTransfer(MTransfer* t);

   void Cancel(int id, MTransfer::MUpdate& u);

   bool SameTransfer(MTransfer* t1, MTransfer* t2);

   MTransferList* LockTransfers() {
      TransferLock->lock();
      return &Transfers;
      }
   void UnlockTransfers() {
      TransferLock->unlock();
      }

   void operator()();

private:
   MTransferList  Transfers;
   boost::mutex::scoped_lock* TransferLock;
   boost::mutex         TransferMutex;
   boost::mutex         StatMutex;


/*
   void           SetActive(bool a) {
      boost::mutex::scoped_lock lock(Mutex);
      Active = a;
      }
*/

   };

//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
} // namespace

#endif

---

MDFS SourceForge project page and download