csclient.cpp

mysee网络直播源代码Mysee Lite是Mysee独立研发的网络视频流媒体播放系统。在应有了P2P技术和一系列先进流媒体技术之后

	SendEnd();
}

void CSClient::SendResList() {
	CriticalSection::Owner lock(sendCS);
	if(!isLogin || !comm->currRes)
		return;
	if(!SendBegin(NP2TS_RES_LIST))
		return;

	// 要共享的资源个数
	UINT8 resNum = 1;
	CopyMoveDes(content, &resNum, sizeof(resNum));

	// 依次写入所有要共享的资源
	CopyMoveDes(content, comm->currRes->GetHashCode().data(), MD5_LEN);
	// send block intervals
	UINT8 intervalNum = MAX_BLOCK_INTERVALS;
	comm->currRes->GetAllIntervals((BlockInterval*)(content+sizeof(intervalNum)), intervalNum);
	CopyMoveDes(content, &intervalNum, sizeof(intervalNum));
	content += intervalNum*sizeof(BlockInterval);

	comm->logFile.StatusOut("CSClient sent NP2TS_RES_LIST.");

	SendEnd();
}

void CSClient::SendReqRes() {
	CriticalSection::Owner lock(sendCS);
	assert(comm->currRes);

	// 如果已经登录，则发送请求资源；否则，登录成功之后再发送
	if(!isLogin) {
		comm->logFile.StatusOut("NOT LOGIN YET, first request will be sent after recving TS2NP_WELCOME...");
		return;
	}
	if(!comm->currRes)
		return;

	if(!SendBegin(NP2TS_REQ_RES))
		return;

	// request resource md5
	CopyMoveDes(content, comm->currRes->GetHashCode().data(), MD5_LEN);

	// send block intervals
	UINT8 intervalNum = MAX_BLOCK_INTERVALS;
	comm->currRes->GetAllIntervals((BlockInterval*)(content+sizeof(intervalNum)), intervalNum);
	CopyMoveDes(content, &intervalNum, sizeof(intervalNum));
	content += intervalNum*sizeof(BlockInterval);

	// current block ID
	UINT currBlockID = comm->currRes->GetPlayingBlock();
	CopyMoveDes(content, &currBlockID, sizeof(currBlockID));

	// need CP?
	bool needCP = (comm->p2pMgr.GetCPConCount(true) < MAX_CONNECTION_OF_CP);
	CopyMoveDes(content, &needCP, sizeof(needCP));

	comm->logFile.StatusOut("CSClient sent NP2TS_REQ_RES.");
	
	SendEnd();

	// 发送了请求资源的消息，等待请求成功的消息返回
	isReqResSucceed = false;
}

void CSClient::SendDelRes(BaseResource* res) {
	CriticalSection::Owner lock(sendCS);
	if(!res || !isLogin)
		return;
	if(!SendBegin(NP2TS_DEL_RES))
		return;

	// request resource md5
	CopyMoveDes(content, res->GetHashCode().data(), MD5_LEN);

	comm->logFile.StatusOut("CSClient sent NP2TS_DEL_RES(%s).", res->GetHashCode().data());
	
	SendEnd();
}

void CSClient::SendReport(bool bRefresh) {
	CriticalSection::Owner lock(sendCS);
	if(!isLogin)
		return;

	if(!SendBegin(NP2TS_REPORT))
		return;

	// 本机信息
	CorePeerInfo thisPeer;
	comm->p2pMgr.GetSelfInfo(thisPeer);
	CopyMoveDes(content, &thisPeer, sizeof(thisPeer));

	// 如果没有当前资源，则发送bRefresh=true和intervalNum=0
	if(!comm->currRes)
		bRefresh = true;

	// 是否更新全部区间列表
	CSClient::CopyMoveDes(content, &bRefresh, sizeof(bRefresh));

	// send block intervals
	// TODO: 这里存在重复代码，有待改进
	UINT8 intervalNum = comm->currRes?MAX_BLOCK_INTERVALS:0;
	if(bRefresh) {
		if(comm->currRes) {
			// 如果是更新全部区间，则取得当前资源全部区间的列表，并发送
			comm->currRes->GetAllIntervals((BlockInterval*)(content+sizeof(intervalNum)), intervalNum);
		}
		CSClient::CopyMoveDes(content, &intervalNum, sizeof(intervalNum));
		content += intervalNum*sizeof(BlockInterval);
		assert(intervalNum <= 20);
	}
	else {
		// 如果是更新变化的区间，则先后写入增加的区间和减少的区间
		comm->currRes->GetDiffIntervals((BlockInterval*)(content+sizeof(intervalNum)), intervalNum, true, true);
		CSClient::CopyMoveDes(content, &intervalNum, sizeof(intervalNum));
		content += intervalNum*sizeof(BlockInterval);
		assert(intervalNum <= 20);

		comm->currRes->GetDiffIntervals((BlockInterval*)(content+sizeof(intervalNum)), intervalNum, true, false);
		CSClient::CopyMoveDes(content, &intervalNum, sizeof(intervalNum));
		content += intervalNum*sizeof(BlockInterval);
		assert(intervalNum <= 20);
	}

	// 发送本机传输信息
	TransferInfo ti;
	comm->p2pMgr.GetTransferInfo(ti);
	CSClient::CopyMoveDes(content, &ti, sizeof(TransferInfo));

	comm->logFile.StatusOut("CSClient sent NP2TS_REPORT.");

	SendEnd();
}

void CSClient::SendNeedPeers() {
	CriticalSection::Owner lock(sendCS);
 	if(!comm->currRes || !isLogin)
		return;

	// 如果请求资源还没有成功，就发送请求消息，然后返回
	if(!isReqResSucceed) {
		SendReqRes();
		return;
	}

	// 如果当前下载速度大于直播的码率，则不必发送NeedPeers
	if(comm->p2pMgr.GetCurDownSpeed()/1024 > comm->currRes->GetBitRate())
		return;

	// 如果已知的Peer列表足够多，就不用发送NeedPeers
	if(comm->p2pMgr.GetKnownPeerCount() + comm->p2pMgr.GetAllCount() >= MIN_KNOWN_PEERS)
		return;	

	if(!SendBegin(NP2TS_NEED_PEERS))
		return;

	// 本机是否需要CachePeer的地址
	bool needCP = (comm->p2pMgr.GetCPConCount(true) < MAX_CONNECTION_OF_CP);
	CopyMoveDes(content, &needCP, sizeof(needCP));


	UINT currBlockID = comm->currRes->GetPlayingBlock();
	CopyMoveDes(content, &currBlockID, sizeof(currBlockID));

	// 本机在P2P网络中所属的层
	UINT8 layer = comm->p2pMgr.GetSelfLayer();
	CopyMoveDes(content, &layer, sizeof(layer));

	comm->logFile.StatusOut("CSClient sent NP2TS_NEED_PEERS.");
	
	SendEnd();
}

void CSClient::SendQueryRes() {
	// TODO: 什么时候发送请求的，目前客户端不关心现在有多少频道(md5)，每个频道有多少人
}

void CSClient::SendLogout() {
	CriticalSection::Owner lock(sendCS);
 	if(!comm->currRes || !isLogin)
		return;

	if(!SendBegin(NP2TS_LOGOUT))
		return;
	
	SendEnd();

	comm->logFile.StatusOut("CSClient sent NP2TS_LOGOUT.");

	// 退出登录
	isLogin = false;
	loginFailCount = 0;
}

void CSClient::SendReport2() {
	CriticalSection::Owner lock(sendCS);
	if(!comm->currRes || !isLogin)
		return;

	if(!SendBegin(NP2TS_REPORT2))
		return;
	
	UINT16 value = 0;

	// playing block
	UINT currBlockID = comm->currRes->GetPlayingBlock();
	CopyMoveDes(content, &currBlockID, sizeof(currBlockID));

	// 当前缓冲的时间，没有缓冲的话就是0
	value = comm->currRes->GetBufferingTime();
	CopyMoveDes(content, &value, sizeof(value));

	// 之前缓冲的次数
	value = comm->currRes->GetBufferCount();
	CopyMoveDes(content, &value, sizeof(value));

	// 之前所有缓冲所用的时间
	value = comm->currRes->GetBufferTime();
	CopyMoveDes(content, &value, sizeof(value));

	// connect Fail Count
	value = comm->p2pMgr.GetConnectFailCount();
	CopyMoveDes(content, &value, sizeof(value));

	// incoming connection count
	value = comm->p2pMgr.GetTotalIncomingCount();
	CopyMoveDes(content, &value, sizeof(value));

	// outgoing connection count
	value = comm->p2pMgr.GetTotalOutgoingCount();
	CopyMoveDes(content, &value, sizeof(value));

	// avg incoming connection elapsed time
	value = comm->p2pMgr.GetAvgIncomingTime();
	CopyMoveDes(content, &value, sizeof(value));

	// avg outgoing connection elapsed time
	value = comm->p2pMgr.GetAvgOutgoingTime();
	CopyMoveDes(content, &value, sizeof(value));

	// message percent
	float msgPercent = comm->p2pMgr.GetMessagePercent();
	CopyMoveDes(content, &msgPercent, sizeof(msgPercent));

	// 发送本机传输信息
	TransferInfo ti;
	comm->p2pMgr.GetTransferInfo(ti);
	CSClient::CopyMoveDes(content, &ti, sizeof(TransferInfo));

	comm->logFile.StatusOut("CSClient sent NP2TS_REPORT2.");

	SendEnd();
}

BOOL CSClient::SendBegin(UINT8 msgType) {
	// 先留着消息大小不写，到最后再写
	content = recvBuf+sizeof(UINT);

	// 消息类型
	CopyMoveDes(content, &msgType, sizeof(msgType));

	if(msgType != NP2TS_LOGIN) {
		// TS的验证码
		CopyMoveDes(content, checkCode, 7);
	}

	return TRUE;
}

void CSClient::SendEnd() {
	// 消息的大小就是移动的指针减去初始的指针
	UINT msgSize = content-recvBuf;
	memcpy(recvBuf, &msgSize, sizeof(msgSize));

	// 发送
	sockaddr_in addr;
	memcpy(&addr, &comm->trackerIP, sizeof(comm->trackerIP));
	
	int ret = sendto(m_Socket, (const char*)recvBuf, msgSize, 
		0, (SOCKADDR*)&addr, sizeof(addr));
	if(ret == SOCKET_ERROR) {
		comm->logFile.StatusErr("sendto() error, rebind udp socket", WSAGetLastError());
		// 重新连接TS
		Reinit();
	}
}

void CSClient::CopyMoveSrc(void * des, const char *& src, size_t size) {
	assert(des && src);
	if(!des || !src)
		return;
	memcpy(des, src, size);
	src += size;
}

void CSClient::CopyMoveDes(char *& des, const void * src, size_t size) {
	assert(des && src);
	if(!des || !src)
		return;
	memcpy(des, src, size);
	des += size;
}

UINT8 CSClient::GetLocalIPList(in_addr*& addrList) {
	UINT8 ipCount = 0;
	addrList = NULL;

	// Get local host name
	char szHostName[128] = "";
	if(gethostname(szHostName, sizeof(szHostName)))
		return 0;

	// Get local IP addresses
	hostent *pHost = 0;
	if((pHost = gethostbyname(szHostName)) == NULL)
		return 0;

	// 计算IP地址个数
	while(pHost->h_addr_list[ipCount]) {
		ipCount++;
		// 注意不能越界了，超过0xff个IP地址将不被支持
		if(ipCount == 0xff)
			break;
	}

	// 分配数组，要在外部释放
	addrList = new in_addr[ipCount];
	if(!addrList)
		return 0;

	// 复制IP列表
	for(int i = 0; i < ipCount; ++i) {
		memcpy(&addrList[i], pHost->h_addr_list[i], pHost->h_length);
	}

	// 返回IP地址个数
	return ipCount;
}

// 重新初始化
void CSClient::Reinit() {
	if(m_Socket != INVALID_SOCKET)
		closesocket(m_Socket);
	
	Init();

	SendLogin();
}

}