peer.cpp

缓冲 缓冲 缓冲 缓冲 缓冲 缓冲

// Peer.cpp: implementation of the Peer class.
//
//////////////////////////////////////////////////////////////////////

#include "stdafx.h"

#include "Peer.h"
#include <stdio.h>
#include <string.h>
#ifdef _WIN32
#include <Process.h>
#endif

#ifdef _DEBUG
#undef THIS_FILE
static char THIS_FILE[]=__FILE__;
#define new DEBUG_NEW
#endif

#define MAXLINUELEN 1024
#define MAXFILENAMELEN 1024

//////////////////////////////////////////////////////////////////////
// Construction/Destruction
//////////////////////////////////////////////////////////////////////

/*
 * read from file the list of neighbor peer info
 * build neighbor lists
 * neighbor file format is as follows:
 * <Neighbor ID>	<Neighbor HostName>	<Neighbor ServPort>	
 * one line for one neighbor peer,each data coloumn is separated by tab
 */

typedef struct
{
	char str[MAXLINE];
	int fd;
}tparam;

static HANDLE hEvent;

Peer::Peer()
{
}

Peer::~Peer()
{
}

static unsigned int _stdcall ThreadDownLoadFunc(void *pVoid)//file->socket
{
	char fname[MAXFILENAMELEN];
	tparam* tp = (tparam *)pVoid;
	int sockfd = tp->fd;
	strncpy(fname,&tp->str[0],sizeof(fname));
	fname[sizeof(fname)-1]='\0';
	delete tp;
	DiskFile inputFile(fname,"rb");
	Socket outputFile;
	outputFile.setSock(sockfd);

	if(inputFile.getStatus()==CLOSED||sockfd==INVALID_SOCKET)
		return 0;
	
	MultiBuffer multiBuffer(&inputFile,&outputFile);
	multiBuffer.start();
	return 1;
}

static unsigned int _stdcall ThreadUpLoadFunc(void *pVoid)//socket->file
{
	char fname[MAXFILENAMELEN];
	tparam* tp = (tparam *)pVoid;
	int sockfd = tp->fd;
	strncpy(fname,&tp->str[0],sizeof(fname));
	fname[sizeof(fname)-1]='\0';
	delete tp;
	DiskFile outputFile(fname,"wb");
	Socket inputFile;
	inputFile.setSock(sockfd);

	if(outputFile.getStatus()==CLOSED||sockfd==INVALID_SOCKET)
		return 0;
	
	MultiBuffer multiBuffer(&inputFile,&outputFile);
	multiBuffer.start();
	return 1;
}

static unsigned int _stdcall ThreadServFunc(void *pVoid)
{
	Peer *p = (Peer*)pVoid;
	if(!p)
		return 0;
	p->startSelect();
	return 0;
}

void Peer::startCommand()
{
	char buf[MAXLINE];
	char *neighborID,*load,*f1,*f2;
	const char *seps = " ";
    while(1)
	{
         fgets(buf,sizeof(buf),stdin);
		 buf[strlen(buf)-1]='\0';
		 load = strtok(buf,seps);
		 if(!strcmp(load,"download"))
		 {
			 neighborID = strtok(NULL,seps);
			 string id(neighborID);
			 Ptr pinfo = getPeerInfo(id);
			 if(!pinfo)
			 {
				 printf("Unknown Neighbor\n");
				 continue;
			 }
			 Socket s;
			 bool isConnected = connect(s,*pinfo);
			 if(!isConnected)
			 {
				 printf("Could not connect to server:%s\n",pinfo->getAddress().c_str());
				 continue;
			 }
			 int sockfd = s.getSock();
			 char wbuf[MAXLINE];
			 wbuf[0] = DOWNLOAD;
			 f1 = strtok(NULL,seps);
			 wbuf[1]='\0';
			 strcat(&wbuf[1],f1);
			 wbuf[sizeof(wbuf)-1]='\0';

			 //negotiate with peer
	         if(s.write(wbuf,strlen(wbuf)+1)!=SOCKET_ERROR)
			 {
				 wbuf[0] = '0';
				 if(s.read(wbuf,sizeof(wbuf))==SOCKET_ERROR||wbuf[0]=='0')
				 {
					 //error
					 printf("Download File %s error,no such file exists\n",f1);
					 s.close();
					 continue;
				 }
				 tparam* tp = new tparam;
				 tp->fd = sockfd;
				 f2 = strtok(NULL,seps);
				 string local_File(f2);
				 local_File+=PATH_SEPARATOR_STR;
				 char *p;
				 for(p = f1+strlen(f1)-1;p>=f1&&*p!=PATH_SEPARATOR;p--);
				 if(*p==PATH_SEPARATOR)p++;
				 local_File.append(p);
				 tp->str[0] = STRING_END;
				 strcat(tp ->str,local_File.c_str());
				 tp->str[MAXLINE-1]=STRING_END;

				 HANDLE threadHandle;
	             unsigned threadID;
				 threadHandle = (HANDLE)_beginthreadex(NULL,
		                                        0,
												ThreadUpLoadFunc,
                                                tp,
                                                0,
                                                &threadID
												);
				 CloseHandle(threadHandle);
			 }else
			 {
				 s.close();
			 }
		 }/*if(!strcmp(load,"download"))*/
		 else if(!strcmp(load,"upload"))
		 {
			 neighborID = strtok(NULL,seps);
			 string id(neighborID);
			 Ptr pinfo = getPeerInfo(id);
			 if(!pinfo)
			 {
				 printf("Unknown Neighbor\n");
				 continue;
			 }
			 Socket s;
			 bool isConnected = connect(s,*pinfo);
			 if(!isConnected)
			 {
				 printf("Could not connect to server:%s\n",pinfo->getAddress().c_str());
				 continue;
			 }
			 int sockfd = s.getSock();
			 char wbuf[MAXLINE];
			 wbuf[0] = UPLOAD;
			 f2 = strtok(NULL,seps);
			 f1 = strtok(NULL,seps);
			 string remote_File(f1);
			 char *p;
			 remote_File+=PATH_SEPARATOR_STR;
			 for(p = f2+strlen(f2)-1;p>=f2&&*p!=PATH_SEPARATOR;p--);
			 if(*p==PATH_SEPARATOR)p++;
			 remote_File.append(p);

			 wbuf[1] = STRING_END;
			 strcat(&wbuf[1],remote_File.c_str());
			 wbuf[sizeof(wbuf)-1]=STRING_END;
			 if(s.write(wbuf,strlen(wbuf)+1)!=SOCKET_ERROR)
			 {
				 wbuf[0] = '0';
				 if(s.read(wbuf,sizeof(wbuf))==SOCKET_ERROR||wbuf[0]=='0')
				 {
					 //error
					 printf("Upload File %s error,remote directory does not exist or is not writable\n",remote_File.c_str());
					 s.close();
					 continue;
				 }
				 tparam* tp = new tparam;
				 tp->fd = sockfd;
				 strcpy(tp ->str,f2);
				 tp->str[MAXLINE-1]='\0';
				 HANDLE threadHandle;
				 unsigned threadID;
				 threadHandle = (HANDLE)_beginthreadex(NULL,
		                                        0,
												ThreadDownLoadFunc,
                                                tp,
                                                0,
                                                &threadID
												);
				 CloseHandle(threadHandle);
			 }else
			 {
				 s.close();
			 }
		 }/*else if(!strcmp(load,"upload"))*/
		 else if(!strcmp(load,"quit"))
		 {
			 offline();
			 printf("Quit!\n");
			 break;
		 }else
		 {
			 printf("Illegal command\n");
			 break;
		 }
	 }/*end while*/

}

void Peer::init(const char *configFile)
{
	FILE *fp;
	
    fp = fopen(configFile,"r");
	if(!fp)
	{
		flag = UNINIT;
	}else
	{
		char buf[MAXLINUELEN];

		while(!feof(fp)&&fgets(buf,sizeof(buf),fp)!=NULL)
		{
			char* pid=0;
			char *addr=0;
			unsigned short portNum=0;
			char *tdir=0;
            char *seps = "\t";
			char *token = strtok(buf,seps);
			if(token!=NULL)
			{
				pid = token;
				token = strtok(NULL,seps);
				if(token!=NULL)
				{
					addr = token;
                    token = strtok(NULL,seps);
					if(token!=NULL)
					{
                       sscanf(token,"%u",&portNum);
					}
				}
			}
			Ptr peerInfo = new PeerInfo(pid,addr,portNum);
            neighbors.push_back(peerInfo);
		}/*end while*/
		
		//start server
		hEvent = CreateEvent(NULL,FALSE,FALSE,NULL);
		HANDLE threadHandle;
		unsigned threadID;
		threadHandle = (HANDLE)_beginthreadex(NULL,
		                               0,
									   ThreadServFunc,
                                       this,
                                       0,
                                       &threadID
									  );
	    printf("commands:\n");
	    printf("download <neighbor id> <remotefile> <local_dir>\n");
	    printf("upload <neighbor id> <local_file> <remote_dir>\n");
		startCommand();
		CloseHandle(threadHandle); //close serv thread handle
	}
}

//connect to central server
void Peer::connect()
{
}

//disconnect from central server
void Peer::disconnect(bool graceless)
{
}

//connect to peer
bool Peer::connect(Socket &s,const PeerInfo& peer)
{
	s.create(TYPE_TCP);
	s.connect(peer.getAddress(),peer.getPort());
	return s.isConnected();
}

PeerInfo& Peer::info()
{
	return self_info;
}
    
int Peer::getNeighborCount() const
{
	return neighbors.size();
}

void Peer::connectToServer(const string &server_addr,const unsigned int &port)
{

}

void Peer::registerToServer()
{

}

Ptr Peer::getPeerInfo(const string &peer_id)
{
	for(vector<Ptr>::iterator iter = neighbors.begin();iter!=neighbors.end(); ++iter)
		if((*iter)->getId()==peer_id)
			return *iter;
	return NULL;
} 


string Peer::getLocalIp() const
{
	string ip;
	return ip;
}

void Peer::online()
{
	establish();
	flag = ONLINE;
}
void Peer::offline()
{
	flag = OFFLINE;
	while(WaitForSingleObject(hEvent,INFINITE)!=WAIT_OBJECT_0);
}



void Peer::startSelect()
{
	int i,maxi,maxfd,listenfd,connfd,sockfd;
    int nready,client[FD_SETSIZE];
	int n;
	char buf[MAXLINE];
    socklen_t clilen;
	struct sockaddr_in cliaddr;
	fd_set allset,rset;

	online();
	listenfd=maxfd = listenSocket.getSock();
	maxi = -1;

	for(i=0;i<FD_SETSIZE;i++)
		client[i] = -1;
	FD_ZERO(&allset);
	FD_SET(listenfd,&allset);
    struct timeval timeout;
	timeout.tv_sec = 0;
	timeout.tv_usec = 200; //timeout = 200ms
	while(flag==ONLINE)
	{
		rset = allset;
		nready = select(maxfd+1,&rset,NULL,NULL,&timeout);
		if(nready==SOCKET_ERROR)
		{
#ifdef WIN32
			printf("Peer serv thread select error:%d\n",::WSAGetLastError());
#else
			printf("Peer serv thread select error:%d\n",errno);
#endif
	        flag = OFFLINE;
			break;
		}
		if(nready==0)
		{
			//timeout
			continue;
		}
		if(FD_ISSET(listenfd,&rset))
		{
			clilen = sizeof(cliaddr);
			connfd = accept(listenfd,(struct sockaddr *)&cliaddr,&clilen);
			for(i=0;i<FD_SETSIZE;i++)
				if(client[i]<0)
				{
				   client[i] = connfd;
				   break;
				}
			if(i==FD_SETSIZE)
				break;
			FD_SET(connfd,&allset);
			if(connfd>maxfd)
				maxfd = connfd;
			if(i>maxi)
				maxi = i;
			if(--nready<=0)
				continue;
		}

		
		for(i=0;i<=maxi;i++)
		{
			if((sockfd=client[i])<0)
				continue;
			if(FD_ISSET(sockfd,&rset))
			{
				if((n=recv(sockfd,buf,sizeof(buf),0))==0)
				{
					closesocket(sockfd);
					FD_CLR(sockfd,&allset);
					client[i]= - 1;
				}else{
					FD_CLR(sockfd,&allset);
					client[i]=-1;
					if(buf[0]==DOWNLOAD)
					{
						/*download file->socket*/
					   tparam* tp = new tparam;
					   tp->fd = sockfd;
					   strncpy(tp ->str,&buf[1],MAXLINE-1);
					   tp->str[MAXLINE-1]='\0';
					   //test if file exists on the peer
					   FILE *testExistFile =fopen(tp->str,"rb");
					   Socket s;
					   s.setSock(tp->fd);
					   if(!testExistFile)
					   {
						   //file does not exist,send response to client
						   char buf[]={'0'};
						   s.write(buf,sizeof(buf));
						   s.close();//close socket
						   continue;
					   }else
					   {
						   char buf[] = {'1'};
						   s.write(buf,sizeof(buf));
					   }

					   fclose(testExistFile);

					   HANDLE threadHandle;
	                   unsigned threadID;
					   threadHandle = (HANDLE)_beginthreadex(NULL,
		                                        0,
												ThreadDownLoadFunc,
                                                tp,
                                                0,
                                                &threadID
												);
					   CloseHandle(threadHandle); //!close download thread handle
					}else if(buf[0]==UPLOAD)
					{
					   /*upload socket->file*/
					   tparam* tp = new tparam;
					   tp->fd = sockfd;
					   strncpy(tp ->str,&buf[1],MAXLINE-1);
					   tp->str[MAXLINE-1]='\0';
					   //test if the file could be uploaded
					   FILE *testExistFile =fopen(tp->str,"wb");
					   Socket s;
					   s.setSock(tp->fd);
					   if(!testExistFile)
					   {
						   char buf[]={'0'};
						   s.write(buf,sizeof(buf));
						   s.close();
						   continue;
					   }else
					   {
						   char buf[] = {'1'};
						   s.write(buf,sizeof(buf));
					   }
					   fclose(testExistFile);

					   HANDLE threadHandle;
	                   unsigned threadID;
					   threadHandle = (HANDLE)_beginthreadex(NULL,
		                                        0,
												ThreadUpLoadFunc,
                                                tp,
                                                0,
                                                &threadID
												);
					   CloseHandle(threadHandle); //!close upload thread handle
					}/*if(buf[0]==DOWNLOAD)*/
				}/*if((n=recv(sockfd,buf,sizeof(buf),0))==0)*/
			}/*if(FD_ISSET(sockfd,&rset))*/
		}/*for(i=0;i<=maxi;i++)*/
	}/*while(flag==ONLINE)*/
	SetEvent(hEvent);
}