communicator.cpp

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
 *
 */
#include "stdafx.h"
#include "Communicator.h"
#include "ConfigFile.h"
namespace NPLayer1 {

Communicator::Communicator() : currRes(NULL), //logFile("#Layer1"), 
		csClient(this), tryClient(this), p2pMgr(this), localPeerFinder(this)
{
	cfgData.COMMUNICATOR_VERSION = 0.152f;
	cfgData.ADD_PROGNAME_VERSION = 0.150f;
	cfgData.ADD_PROGTIME_VERSION = 0.152f;
	cfgData.ACCEPT_VERSION = 0.135f;
	cfgData.OUTGOING_NUM = 30;
	cfgData.INCOMING_NUM = 30;
	cfgData.OUT_4FREE_NUM = 5;
	cfgData.IN_4FREE_NUM = 5;
	cfgData.allow_post = 0;

	noitfyWindow = NULL;
	notifyCode = 0;
	userID = 0;
	userPass[MD5_LEN] = 0;
	
	// 初始化随机数生成器
	ctx.randa=ctx.randb=ctx.randc=(ub4)0;
	// 设置随机种子
	for (int i = 0; i < RANDSIZ; ++i) 
		ctx.randrsl[i] = (ub4)i;
	ctx.randrsl[0] = GetTickCount();
	ctx.randrsl[1] = time(NULL);
	std::random_shuffle(ctx.randrsl, ctx.randrsl+RANDSIZ);
	randinit(&ctx, TRUE);
}

Communicator::~Communicator() {
	logFile.StatusOut("P2PMgr Uninit()...");
	p2pMgr.Uninit();

	logFile.StatusOut("CSClient SendLogout()...");
	csClient.SendLogout();

	StopCurrentRes();

	TE_CleanupLibrary();

	//MessageBox(NULL, logFile.file_name().c_str(), allow_post==1? "1": "0", MB_OK);
	if(cfgData.allow_post) {
		STARTUPINFO si;
		PROCESS_INFORMATION pi;

		ZeroMemory( &si, sizeof(si) );
		si.cb = sizeof(si);
		ZeroMemory( &pi, sizeof(pi) );

		//发送log文件
		TCHAR exename[MAX_PATH_EX] = _T("");
		if(cfgData.module_base.size() + 20 <= MAX_PATH_EX)
			_stprintf(exename, "%sPostLog.exe", cfgData.module_base.data());

		TCHAR cmdarg[MAX_PATH_EX] = _T("");
		if(cfgData.post_log_url.size() + logFile.file_name().size() + 20 <= MAX_PATH_EX)
			_stprintf(cmdarg, _T(" %s %s "), cfgData.post_log_url.data(), logFile.file_name().data());

		if( CreateProcess(  // No module name (use command line). 
			exename, 
			cmdarg,
			NULL,             // Process handle not inheritable. 
			NULL,             // Thread handle not inheritable. 
			FALSE,            // Set handle inheritance to FALSE. 
			0,                // No creation flags. 
			NULL,             // Use parent's environment block. 
			NULL,             // Use parent's starting directory. 
			&si,              // Pointer to STARTUPINFO structure.
			&pi )             // Pointer to PROCESS_INFORMATION structure.
			) 
		{
			// Close process and thread handles. 
			CloseHandle( pi.hProcess );
			CloseHandle( pi.hThread );
		}
	}
	// delete log file
	logFile.Uninit();
}

P2P_RETURN_TYPE Communicator::Init(string trackerURL) {
	if(trackerIP.sin_addr.s_addr != 0)
		return PRT_DUP_INIT;

	TCHAR buf[MAX_PATH_EX] = _T("");
	if(!BufferMgr::GetSelfModulePath(buf, MAX_PATH_EX)) {
		assert(0);
		return PRT_SYS;
	}
	cfgData.module_base = buf;

	// 读取配置文件
	ConfigFile config(cfgData.module_base+"config.ini");
	cfgData.allow_post = atoi(config.Value("postlog", "allowpost").c_str());
	cfgData.post_log_url = config.Value("postlog", "posturl");

	// 初始化日志文件
	P2P_RETURN_TYPE ret = logFile.Init(cfgData.module_base);
	//MessageBox(NULL, module_base.c_str(), "Communicator::Init", MB_OK);
	if(ret < PRT_OK)
		return ret;

	// Try calling GetVersionEx using the OSVERSIONINFOEX structure.
	// If that fails, try using the OSVERSIONINFO structure.
	ZeroMemory(&osvi, sizeof(OSVERSIONINFOEX));
	osvi.dwOSVersionInfoSize = sizeof(OSVERSIONINFOEX);
	if(!GetVersionEx ((OSVERSIONINFO *) &osvi)) {
		osvi.dwOSVersionInfoSize = sizeof (OSVERSIONINFO);
		if(!GetVersionEx ((OSVERSIONINFO*) &osvi)) {
			logFile.StatusErr("无法检测Windows版本,程序必须结束.", GetLastError());
			return PRT_SYS;
		}
	}

	// 初始化Windows Socket
	logFile.StatusOut("Initializing Windows Socket...");
	if(TE_InitLibrary() != SOCKET_SUCCESS)
		return PRT_INIT_SOCK;

	// 取得TrackServer的IP和Port
	int index = trackerURL.find(':');
	if(index != -1) {
		trackerIP.sin_addr.s_addr = TE_GetIP(trackerURL.substr(0, index).data(), TRUE);
		if(trackerIP.sin_addr.s_addr == INADDR_NONE)
			return PRT_TRACKER_IP;
		int port = atoi(trackerURL.substr(index+1, trackerURL.length()-index-1).data());
		if(port == 0)
			return PRT_TRACKER_IP;
		trackerIP.sin_port = htons(port);
	}
	else {
		return PRT_TRACKER_IP;
	}
	logFile.StatusOut("TrackServer %s", trackerURL.data());

	// 删除原来的缓冲文件和日志文件
	LogMgr::RemoveOldTmpFile("GB", cfgData.module_base.data());
	LogMgr::RemoveOldTmpFile("2006", cfgData.module_base.data());

	// 建立新的缓冲文件
	logFile.StatusOut("Initialize Default BufferMgr...");
	bool bFileModified = false;
	ret = g_bufferMgr.Init("", bFileModified, rand(&ctx));
	if(ret < PRT_OK)
		return ret;

	logFile.StatusOut("CSClient Init & SendLogin...");
	ret = csClient.Init();
	if(ret < PRT_OK)
		return ret;

	// 初始化P2P连接，同时也启动了主线程
	logFile.StatusOut("P2PMgr Init...");
	ret = p2pMgr.Init();
	if(ret < PRT_OK)
		return ret;

	// 登陆TrackerServer
	csClient.SendLogin();

	// 初始化Peer发现模块
	localPeerFinder.Init();
	return PRT_OK;
}

P2P_RETURN_TYPE Communicator::Request(
						   const char* resname,			// 资源名
						   const char* hashcode,		// 资源的Hash码
						   const char* strSPList,		// SuperPeer列表的IP地址
						   float bitRate				// 频道的码率
						   )
{
	if(!resname || !hashcode)
		return PRT_WRONG_ARG;

	CriticalSection::Owner lock(mainThreadLock);
	logFile.StatusOut("Request Resource %s, name %s, bitrate %.2f, SP List %s", hashcode, resname, bitRate, strSPList);

	// 停止当前下载的资源
	StopCurrentRes();

	LiveResource* res = new LiveResource(this);
	if(!res)
		return PRT_ALLOC;
	// 使用参数初始化资源
	P2P_RETURN_TYPE ret = res->Init(resname, hashcode, strSPList?strSPList:"", bitRate);
	if(ret < PRT_OK) {
		// 初始化失败
		logFile.StatusErr("Init exist Res", GetLastError());
		assert(0);
		// 删除未使用的资源
		delete res;
		return ret;
	}

	// 初始化成功, 设置当前资源
	currRes = res;

	// 向TS请求资源
	csClient.SendReqRes();

	return PRT_OK;
}

void Communicator::StopCurrentRes() {
	CriticalSection::Owner lock(mainThreadLock);
	// 如果存在当前下载文件, 停止当前文件的下载
	if(currRes) {
		logFile.StatusOut("Stopping connections on current resource...");
		p2pMgr.StopAll();
		/// 在这里停止主线程的话，会需要几秒钟时间，不想让stop太慢，所以不在这里执行
		//p2pMgr.Uninit();
		localPeerFinder.Uninit();

		logFile.StatusOut("CSClient SendDelRes()...");
		csClient.SendDelRes(currRes);
		delete currRes;

		// Log out
		csClient.SendLogout();

		currRes = NULL;
	}
}

// 发送错误消息给外部程序
void Communicator::PostErrMessage(P2P_NOTIFY_TYPE type, LPARAM lParam, bool shouldStop) {

	if(noitfyWindow) {
		WPARAM wparam = MAKEWPARAM((UINT16) type, channelID);
		PostMessage(noitfyWindow, notifyCode, wparam, lParam);
	}
	if(shouldStop) {
		logFile.StatusOut("Post Err Message with code %d.", type);
		StopCurrentRes();
	}
}

}