p2pclient.h

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

/*
 *  Openmysee
 *
 *  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.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program; if not, write to the Free Software
 *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 *
 */
#pragma once

namespace NPLayer1 {

class Communicator;

enum {
	// P2P 缓冲区的大小,比BLOCK稍大
	P2P_BUF_SIZE = BLOCK_SIZE+1024, 
};

// TCP传输的数据包
class TCPPacket {
public:
	TCPPacket() : size(0), sent(0) {};
	~TCPPacket() {};
	char GetMsgType() {return buf[4];};
	UINT* GetBlockID() {return reinterpret_cast<UINT*>(buf+5);};
	UINT* GetBlockSize() {return reinterpret_cast<UINT*>(buf+9);};
	void Init() {};
	void Uninit() {};
	char buf[P2P_BUF_SIZE];
	// 要发送的消息大小
	int size;
	// 已经发送的大小
	int sent;
};

enum MSG_STATE {
	MSG_COMPLETE, MSG_UNCOMPLETE, MSG_ERR_SIZE, MSG_ERR_TYPE, 
	MSG_DIFF_CHNL, MSG_NOMORE_CONS, MSG_ERR_LIST_SIZE, 
	MSG_SEND_ERR, MSG_SAVEDATA_ERR, MSG_UNMATCH_BLOCKID, 
	MSG_NOSUCH_RES_HERE, MSG_REMOTE_ERR, MSG_CHNL_CLOSED, 
	MSG_NOSUCH_RES_SP, MSG_CHNL_END, MSG_LOW_VERSION
};

// 正在连接的Peer
class P2PClient : public TransferCalculator {
public:
	P2PClient();
	virtual ~P2PClient();

	bool IsValid() { return valid;};
	BOOL SetValid(Communicator*, const ConnectingPeer& peer);
	void SetInvalid();

	BOOL SendHello();
	BOOL SendSPUpdate(const SPUpdate& spUpdate, UINT8 selfLayer, BYTE sum);
	BOOL SendReport(const CorePeerInfo&, bool bRefresh);
	BOOL SendNearPeers();
	BOOL SendReqMedia(UINT blockID);

	BOOL IsIdleTooLong();
	BOOL BaseRecv();

    // 清除发送媒体类型区间的记录
    void ClearSentMediaArray() {sentMediaArray.Clear();};

	// 发送push列表中的第一个块
	BOOL SendFirstPushBlock();

	// 查找连接对方是否有此块
	bool FindRemoteBlock(const UINT blockID) const {
		return remoteInterval.FindBlock(blockID);
	};

	bool AddPushBlock(UINT blockID, bool bRefreshing);	// 添加一个PushList中的块
	bool FindPushBlock(const UINT blockID) const;		// 查找一个PushList中的块
	bool DelPushBlock(UINT blockID);					// 删除一个PushList中的块
	bool IsPushListFull();								// Push List是否已满
	UINT8 GetTotalPushSize();							// 取得Pushlist大小
	void ClearPush();									// 清空PushList
	void SendPush();									// 发送整个PushList，因为之前重新分配了所有连接的块
	void ReloadPushList();								// 重新加载PushList，只有在普通的下载时才会被调用

	// 获取第一个Packet进行发送
	void GetPacket(TCPPacket*& packet) {
		if(m_sendList.empty()) {
			packet = NULL;
		}
		else {
			packet = m_sendList.front();
			m_sendList.pop_front();
		}
	};
	// 将Packet放回首部
	void PutPacketBack(TCPPacket* packet) {
		if(packet)
			m_sendList.push_front(packet);
	};

	P2PAddress GetAddress() {return remotePeer;};
	PeerInfoWithAddr GetPeerInfo() {return remotePeer;};
	SOCKET GetSocket() {return m_Socket;};
	bool GetIsIncoming() {return isIncoming;};
	bool GetIsForFree() {return isForFree;};
	bool GetIsCachePeer() {return remotePeer.isCachePeer;};
	bool GetIsSameLan() {return isSameLan;};
	bool GetIsSameRes() {return isSameRes;};
	DWORD GetElapsedTime() {return timeGetTime()-connectionBeginTime;};
	UINT8 GetLayer() {
		if(!valid)
			return 0xff;
		// 避免从0xff回到0
		if(remotePeer.layer != 0xff)
			return static_cast<UINT8>(remotePeer.layer)+1;
		return remotePeer.layer;
	};
	double GetBandWidth() {
		if(!valid || m_transUsedTime == 0)
			return 0.0f;
		return (static_cast<double>(totalDownBytes/1024)/m_transUsedTime)*1000;// 单位 kilo bytes per second
	};

private:
	// 解析消息
	MSG_STATE ParseMsg();

	// 发送消息的函数
	BOOL SendPushList(UINT blockID, bool bAdd);
	BOOL SendResponse(UINT blockID, bool tryGetData);
	BOOL SendMsg();
	BOOL SendMediaType(UINT blockID);

	// 响应消息的函数
	MSG_STATE OnHello();
	MSG_STATE OnSPUpdate();
	MSG_STATE OnReport();
	MSG_STATE OnNearPeers();
	MSG_STATE OnPushList();
	MSG_STATE OnResponse();
	MSG_STATE OnMsg();
	MSG_STATE OnReqMedia();
	MSG_STATE OnMediaType();

	// 每次编制发送消息时，第一步的操作
	BOOL SendBegin(TCPPacket*& packet, UINT8 msgType);
	// 每次编制发送消息时，最后一步的操作
	void SendEnd(TCPPacket*& packet);

	// 检查发送列表中是否已经有准备发送的块
	// 1. 如果没有，那么返回FALSE；
	// 2. 如果有正在发送的块，返回TRUE
	BOOL CheckSendingBlock();

	// 从发送列表中查找一个块，如果此块准备发送（packet->send == 0），则删除并返回TRUE；
	// 如果此块已经开始发送，则直接返回TRUE。如未找到，则返回FALSE
	// 如果blockID是UINT_MAX，则删除任意一个准备发送的块
	BOOL DeleteSendingBlock(UINT blockID);

private:
	enum {
		// 连上CP以后，如果一段时间内没有收到第一条消息，就断开连接
		MAX_CP_FIRST_MSG_IDLE = 10000,
		// 连上NP以后，如果一段时间内没有收到第一条消息，就断开连接
		MAX_NP_FIRST_MSG_IDLE = 10000,
		// 连入连接如果一段时间内不发送数据，就会被断开
		MAX_INCOMING_SENDDATA_IDLE = 25000, 
		// 连出连接如果一段时间内不接受数据，就会被断开
		MAX_OUTGOING_RECVDATA_IDLE = 25000, 
		// 如果push list非空，但是一段时间却没有收到数据头部，就断开连接
		MAX_RESPONSE_IDLE = 10000,

		MAX_P2P_ERRMSG = 64, 
	};
	enum {
		P2P_LOW_PUSH = 1,   // 给较慢( < bitRate/6)的连接每次分配的块数
		P2P_MID_PUSH = 3,	// 给中等( < bitRate/2)的连接每次分配的块数
		P2P_HIGH_PUSH = 5,  // 给较快( > birRate/2)的连接每次分配的块数
	};
	SOCKET	m_Socket;				// 此连接的SOCKET，初始值是INVALID_SOCKET
	char	recvBuf[P2P_BUF_SIZE];	// 接收数据的缓冲区
	UINT	recvOff;				// 缓冲区中数据的长度
	char	*recvPointer;			// 从接收到的数据包读取数据的移动指针
	char	*sendPointer;			// 向被发送的数据包写入数据的移动指针
	char	errStr[MAX_P2P_ERRMSG];	// 出错信息
	UINT	msgSize;

	PeerInfoWithAddr	remotePeer;		// 当前连接Peer的简单信息和IP地址
	IntervalArray		remoteInterval;	// 当前连接Peer所拥有块的区间列表
	PushList			remotePush;		// 需要对方发给本方数据的Push List
	PushList			localPush;		// 需要本方发给对方数据的Push List
	MediaArray			sentMediaArray;	// 曾经发送过媒体类型的的区间
	bool				isIncoming;		// 是否连入连接
	bool				isForFree;		// 是否free连接
	bool				isPassive;		// 是否被动连接
	bool				isSameRes;		// 双方当前资源是否相同
	bool				isSameLan;		// 双方是否相同子网
	float				remoteVersion;	// 对方的版本
	
	bool				bGotFirstMsg;	// 是否已经收到第一条消息
	bool				valid;			// 是否有效连接

	// 用于计算Idle时间，单位是毫秒
	DWORD				lastSendDataTime;		// 上次发送数据的时间
	DWORD				lastRecvDataTime;		// 上次接收数据的时间
	DWORD				lastRecvDataHdTime;		// 上次接收到数据头的时间

	DWORD				lastRequestStartTime;	// 最近一次请求开始的时间
	DWORD				lastRequestBytes;		// 最近一次请求的数据大小
	DWORD				connectionBeginTime;	// 本次连接开始的时间

	// 为了统计真实的传输速度，需要知道传输数据的具体时间
	// 又因为传输是pushlist而非一个接一个的方式，所以需要统计两种时间段：
	// 一种是remotePush从0变成非零的时刻起，收到第一个Block的数据止
	// 一种是收到Block的数据起，收到下一个Block的数据止。
	// 两种情况不会并发进行，所以可以用一个变量进行记录。
	DWORD				m_reqStartTime;			// 需要记录的请求起始时间
	DWORD				m_transUsedTime;		// 用在传输数据上的时间
	Communicator		*comm;					// 主类指针

	list<TCPPacket*>			m_sendList;		// 即将被发送的Packet列表

	typedef list<TCPPacket*>::iterator TCPPackIt;
};
}