#include "stdafx.h"

using namespace std;
using boost::lexical_cast;

struct findFromAddr : public unary_function<TCPSocketBuffer<TimedTCPSocket>, bool> {
	findFromAddr(unsigned long naddr) : naddr_(naddr) {}
	bool operator()(const TCPSocketBuffer<TimedTCPSocket>& s) const
	{
		sockaddr_in sin;
		s.getpeername(reinterpret_cast<sockaddr*>(&sin));
		return sin.sin_addr.s_addr == naddr_;
	}
private:
	unsigned long naddr_;
};

class NodeXMLParser : public XMLParser {
public:
	NodeXMLParser(ChatProcessor* data) : XMLParser()
	{
		XML_SetStartElementHandler(parser, nodeElementStart);
		XML_SetUserData(parser, data);
	}

	virtual ~NodeXMLParser() {}

private:
	static void XMLCALL nodeElementStart(void* data, const char* el, const char** attr);
};

class HandshakeXMLParser : public XMLParser {
public:
	HandshakeXMLParser(ChatProcessor& chatProcessor,
		const TCPSocketBuffer<TimedTCPSocket>& sockbuf)
		: XMLParser(), chatProcessor_(chatProcessor), sockbuf_(sockbuf)
	{
		XML_SetStartElementHandler(parser, elementStart);
		XML_SetUserData(parser, this);
	}
	virtual ~HandshakeXMLParser() {}

private:
	static void XMLCALL elementStart(void* data, const char* el, const char** attr);

	ChatProcessor& chatProcessor_;
	const TCPSocketBuffer<TimedTCPSocket>& sockbuf_;
};

class MessageXMLParser : public XMLParser {
public:
	MessageXMLParser(ChatProcessor& chatProcessor)
		: XMLParser(), ttl(0), chatProcessor_(chatProcessor)
	{
		XML_SetElementHandler(parser, elementStart, elementEnd);
		XML_SetUserData(parser, this);
		XML_SetCharacterDataHandler(parser, charHandler);
	}
	virtual ~MessageXMLParser() {}

private:
	static void XMLCALL elementStart(void* data, const char* el, const char** attr);
	static void XMLCALL elementEnd(void* data, const char* el);
	static void XMLCALL charHandler(void* data, const XML_Char* s, int len);

	ChatProcessor& chatProcessor_;
	string element, guid, name, message, id, sign;
	int ttl;
};

void setChatMessage(const string& name, const string& message, time_t msgtime = 0);
void updateStat(TCPServer<TCPSocketBuffer<TimedTCPSocket>, ChatProcessor>& servent);
string createGUID(void);
string ipaddr2string(unsigned long addr);

namespace {
	//! これらは固定じゃなくて外部から取り込んでもよし
	const int maxInConnection = 16;
	const int maxOutConnection = 4;
	const int maxWaitingConnection = 2;
	const char * const chatNodeFileName = "chatnode.txt";
}

TCPServer<TCPSocketBuffer<TimedTCPSocket>, ChatProcessor>* pchatservent;
extern HWND chatDlg;

unsigned int __stdcall chat_thread(void* /*data*/)
{
	unsigned short chat_port;
	try {
		chat_port = lexical_cast<unsigned short>(getIniString("o2on", "chat_port", ""));
	} catch (boost::bad_lexical_cast&) {
		chat_port = 7744;
	}
	SetDlgItemText(chatDlg, IDC_CHAT_PORT, lexical_cast<string>(chat_port).c_str());

	// ChatProcessor
	ChatProcessor chatprocessor(chat_port);

	// Chat用servent起動
	TCPServer<TCPSocketBuffer<TimedTCPSocket>, ChatProcessor>
		chatservent(chatprocessor, maxInConnection);
	try {
		if (!chatservent.start(chat_port)) {
			MessageBox(NULL, "Chatポートのオープンに失敗しました\n設定を変更して再起動してください", "o2on", MB_OK);
			return 0;
		}
	} catch (exception& e) {
		// もっとちゃんと…
		MessageBox(NULL, e.what(), "o2on", MB_OK);
		return 0;
	}
	pchatservent = &chatservent;
	updateStat(chatservent);

	while (do_thread) {
		chatservent.process();
	}

	return 0;
}

BOOL CALLBACK chat_dialogProc(HWND hwnd, UINT msg, WPARAM wp, LPARAM /*lp*/)
{
	switch (msg) {
	case WM_INITDIALOG:
		{
			HWND msgwnd = GetDlgItem(hwnd, IDC_CHAT_MSG);

			LV_COLUMN lvcol;
			lvcol.mask = LVCF_WIDTH | LVCF_TEXT | LVCF_SUBITEM;
			lvcol.cx = 70;
			lvcol.pszText = "Time";
			lvcol.iSubItem = 0;
			ListView_InsertColumn(msgwnd, 0, &lvcol);

			lvcol.cx = 80;
			lvcol.pszText = "Name";
			lvcol.iSubItem = 1;
			ListView_InsertColumn(msgwnd, 1, &lvcol);

			lvcol.cx = 340;
			lvcol.pszText = "Messages";
			lvcol.iSubItem = 2;
			ListView_InsertColumn(msgwnd, 2, &lvcol);

			MoveWindow(hwnd, 108, 96, 640, 480, TRUE);
		}
		break;

	case WM_SIZE:
		RECT rect;
		GetClientRect(hwnd, &rect);
		MoveWindow(GetDlgItem(hwnd, IDC_CHAT_MSG), 124, 1,
			rect.right - 124 -1, rect.bottom - 21 -2, TRUE);
		MoveWindow(GetDlgItem(hwnd, IDC_CHAT_EDIT), 124, rect.bottom - 21,
			rect.right - 186, 20, TRUE);
		MoveWindow(GetDlgItem(hwnd, IDC_CHAT_NAME), 1, rect.bottom - 21, 122, 20, TRUE);
		MoveWindow(GetDlgItem(hwnd, IDC_CHAT_POST), rect.right - 61, rect.bottom - 21,
			60, 20, TRUE);

		break;

	case WM_COMMAND:
		switch (LOWORD(wp)) {
		case IDC_CHAT_POST:
			{
				char message[256], name[32];
				GetDlgItemText(hwnd, IDC_CHAT_EDIT, message, sizeof message -1);
				GetDlgItemText(hwnd, IDC_CHAT_NAME, name, sizeof name -1);
				if (message[0] != '\0') {
					if (pchatservent) {
						pchatservent->getProcessor().queueChatMessage(name, message);
					}
					SetDlgItemText(hwnd, IDC_CHAT_EDIT, "");
				}
			}
			break;

		case IDC_CHAT_ADDNODE:
			{
				char nodestr[256];
				GetDlgItemText(hwnd, IDC_CHAT_NEWNODE, nodestr, sizeof nodestr -1);
				string buf = nodestr;
				if (!buf.empty()) {
					if (pchatservent) {
						//pchatservent->getProcessor().pushNode(node);
						try {
							size_t pos = buf.rfind(":");
							pchatservent->getProcessor().pushNode(NodeInfo(buf.substr(0, pos),
                                boost::lexical_cast<unsigned short>(buf.substr(pos + 1))));
							// この後でsort, uniqueしたほうがいいかな？
						} catch (exception&) {}
						updateStat(*pchatservent);
					}
					SetDlgItemText(hwnd, IDC_CHAT_NEWNODE, "");
				}
			}
			break;

		case IDC_CHAT_NODEENC:
			{
				const static boost::regex ipaddr("(\\d+)\\.(\\d+)\\.(\\d+)\\.(\\d+)");
				char addrstr[16];
				GetDlgItemText(hwnd, IDC_CHAT_ADDR, addrstr, sizeof addrstr -1);
				string addr(addrstr);
				boost::smatch results;
				if (boost::regex_search(addr, results, ipaddr)) {
					Node node = {0};
					node.ip = lexical_cast<int>(results.str(1))
						| lexical_cast<int>(results.str(2)) << 8
						| lexical_cast<int>(results.str(3)) << 16
						| lexical_cast<int>(results.str(4)) << 24;
					if (pchatservent) {
						node.port = pchatservent->getProcessor().getChatport();
					}
					char buf[32];
					node_enc(&node, buf);
					SetDlgItemText(chatDlg, IDC_CHAT_NODE, buf);
				}
			}
			break;

		default:
			return FALSE;
		}
		break;

	//case WM_PAINT:
	//	{
	//		PAINTSTRUCT ps;
	//		HDC hdc = BeginPaint(hwnd, &ps);
	//		EndPaint(hwnd, &ps);
	//	}
	//	break;
	default:
		return FALSE;
	}
	return TRUE;
}

void setChatMessage(const string &name, const string &message, time_t msgtime)
{
	HWND hlist = GetDlgItem(chatDlg, IDC_CHAT_MSG);
	static int num = -1;
	if (num > 100) {
		ListView_DeleteItem(hlist, 0);
	} else {
		++num;
	}

	using namespace boost::posix_time;
	string timestr(to_simple_string(second_clock::local_time().time_of_day()));
	if (msgtime != 0) {
		time_t now;
		time(&now);
		timestr += "(" + lexical_cast<string>(now - msgtime) + ")";
	}
	LV_ITEM item;
	item.mask = LVIF_TEXT;
	item.pszText = const_cast<char*>(timestr.c_str());
	item.iItem = num;
	item.iSubItem = 0;
	ListView_InsertItem(hlist, &item);

	item.pszText = const_cast<char*>(name.c_str());
	item.iSubItem = 1;
	ListView_SetItem(hlist, &item);

	item.pszText = const_cast<char*>(message.c_str());
	item.iSubItem = 2;
	ListView_SetItem(hlist, &item);

	ListView_Scroll(hlist, 0, 16);
}

void updateStat(TCPServer<TCPSocketBuffer<TimedTCPSocket>, ChatProcessor>& servent)
{
	SetDlgItemText(chatDlg, IDC_ACTIVE_NODE, lexical_cast<string>(servent.numofConnection()).c_str());
	SetDlgItemText(chatDlg, IDC_ALL_NODE, lexical_cast<string>(servent.getProcessor().getNodessize()).c_str());
}

void ChatProcessor::outboundHandshake(const TCPSocket& s, unsigned short myport, unsigned long yourip)
{
	string handshake = boost::str(boost::format(
		"<o2on><protocol name=\"o2on\" ver=\"0.2\"/>"
		"<client name=\"o2on\" ver=\"0.1\"/>"
		"<nodeinfo myport=\"%1%\" yourip=\"%2%\"/></o2on>") % myport % ipaddr2string(yourip));
	//s.send(handshake.data(), handshake.length());
	sendP2PMessage(s, "HANDSHAKE", vector<char>(handshake.begin(), handshake.end()));
}

//bool ChatProcessor::parseProtocol(TCPSocketBuffer<TimedTCPSocket>& s)
//{
//	try {
//		string& buffer = s.getBuffer();
//		do {
//			size_t pos = 7 + buffer.find("</o2on>");
//			string xml(buffer, 0, pos);
//			buffer.erase(0, pos);
//
//			pDoc->loadXML(_bstr_t(xml.c_str()));
//			MSXML2::IXMLDOMNodePtr pRoot(pDoc->selectSingleNode(L"/o2on"));
//
//			map<string, string> attributes;
//			mapXmlAttribute(pRoot, attributes);
//
//			if (attributes["service"] == "system") {
//				if (attributes["type"] == "handshake") {
//					map<string, string> nodeinfo;
//					mapXmlAttribute(pRoot->selectSingleNode(L"nodeinfo"), nodeinfo);
//
//					sockaddr_in sin;
//					s.getpeername(reinterpret_cast<sockaddr *>(&sin));
//					Node node;
//					node.ip = sin.sin_addr.s_addr;
//					try {
//						node.port = lexical_cast<unsigned short>(nodeinfo["myport"]);
//					} catch (boost::bad_lexical_cast&) {
//						node.port = 7744;
//					}
//
//					try {
//						global_ip = lexical_cast<unsigned long>(nodeinfo["yourip"]);
//						string ipaddr(boost::str(boost::format("%d.%d.%d.%d") %
//							(global_ip & 0xff) %
//							((global_ip >> 8) & 0xff) %
//							((global_ip >> 16) & 0xff) %
//							((global_ip >> 24) & 0xff)));
//						SetDlgItemText(chatDlg, IDC_CHAT_ADDR, ipaddr.c_str());
//						Node node;
//						node.ip = global_ip;
//						node.port = chat_port_;
//						char buf[32];
//						node_enc(&node, buf);
//						SetDlgItemText(chatDlg, IDC_CHAT_NODE, buf);
//					} catch (boost::bad_lexical_cast&) {}
//
//					Lock lock(cs_nodes);
//					nodes.push_back(node);
//					sort(nodes.begin(), nodes.end());
//					nodes.erase(unique(nodes.begin(), nodes.end()), nodes.end());
//					updateStat(*server_);
//
//					sendNodes(s.getSocket(), nodes);	// 暫定的にここでもノードを投げる
//				}
//			} else if (attributes["service"] == "chat") {
//				if (attributes["type"] == "message") {
//					string guid(pRoot->selectSingleNode(L"guid")->Gettext());
//					Lock lock(cs_msgq);
//					if (find(processed.begin(), processed.end(), guid) == processed.end()) {
//						// このGUIDを持つメッセージは始めて到達した
//						processed.push_back(guid);
//						if (processed.size() > 100) {
//							processed.pop_front();
//						}
//						string name(pRoot->selectSingleNode(L"name")->Gettext());
//						string message(pRoot->selectSingleNode(L"message")->Gettext());
//						time_t msgtime;
//						try {
//							msgtime	= lexical_cast<time_t>(
//								pRoot->selectSingleNode(L"time")->Gettext());
//						} catch (boost::bad_lexical_cast&) {
//							time(&msgtime);
//						}
//						setChatMessage(name, message, msgtime);
//
//						time(&msgtime);
//						Lock lock2(cs_msgs);
//						messages.push_back(ChatMessage(name, message, guid, msgtime, "", ""));
//						if (messages.size() > 500) {
//							messages.pop_front();
//						}
//
//						// TTLが正ならまわりにもう一回投げる
//						int ttl;
//						try {
//							ttl = lexical_cast<int>(pRoot->selectSingleNode(L"ttl")->Gettext());
//						} catch (boost::bad_lexical_cast&) {
//							ttl = 0;
//						}
//						if (--ttl > 0) {
//							string id(pRoot->selectSingleNode(L"id")->Gettext());
//							string sign(pRoot->selectSingleNode(L"sign")->Gettext());
//							msgqueue.push(ChatMessage(name, message, guid, 0, id, sign, ttl));
//						}
//					}
//				} else if (attributes["type"] == "nodes") {
//					// ノードが飛んできた
//					Lock lock(cs_nodes);
//					MSXML2::IXMLDOMNodeListPtr pnodes(pRoot->selectNodes(L"node"));
//					for (long i = 0; i < pnodes->Getlength(); ++i) {
//						Node node = {0};
//						if (node_dec(pnodes->Getitem(i)->Gettext(), &node)) {
//							nodes.push_back(node);
//						}
//					}
//
//					sort(nodes.begin(), nodes.end());
//					nodes.erase(unique(nodes.begin(), nodes.end()), nodes.end());
//					updateStat(*server_);
//				}
//			}
//		} while (strfound(buffer.find("</o2on>")));
//	} catch (_com_error&) {
//		return false;
//	}
//
//	return true;
//}

//void ChatProcessor::mapXmlAttribute(const MSXML2::IXMLDOMNodePtr& pNode, map<string, string>& dest)
//{
//	MSXML2::IXMLDOMNamedNodeMapPtr pAttr(pNode->Getattributes());
//	for (long i = 0; i < pAttr->length; ++i) {
//		MSXML2::IXMLDOMNodePtr pAttrNode(pAttr->Getitem(i));
//		dest.insert(pair<string, string>(string(pAttrNode->GetnodeName()),
//			string(pAttrNode->Gettext())));
//	}
//
//}

void ChatProcessor::queueChatMessage(const string& name, const string& message)
{
	//time_t now;
	//time(&now);
	Lock lock(cs_msgq);
	msgqueue.push(ChatMessage(name, message, createGUID(), 0, "", ""));
}

// 要検討(w)
string createGUID(void)
{
	GUID guid;
	CoCreateGuid(&guid);

	return string(boost::str(boost::format("%08X-%04X-%04X-%02X%02X-%02X%02X%02X%02X%02X%02X")
		% guid.Data1 % guid.Data2 % guid.Data3 % static_cast<int>(guid.Data4[0])
		% static_cast<int>(guid.Data4[1]) % static_cast<int>(guid.Data4[2])
		% static_cast<int>(guid.Data4[3]) % static_cast<int>(guid.Data4[4])
		% static_cast<int>(guid.Data4[5]) % static_cast<int>(guid.Data4[6])
		% static_cast<int>(guid.Data4[7])));
}

const string ChatMessage::toXmlString() const
{
	int ttl = (ttl_ != -1) ? ttl_ : rand() % 50 + 20;
	return string(str(boost::format("<o2on><name>%1%</name><message>%2%</message>"
		"<guid>%3%</guid><id>%4%</id><sign>%5%</sign><ttl>%6%</ttl></o2on>")
		% name_ % message_ % guid_ % id_ % sign_ % ttl));
}

void ChatMessage::sanitize(string& str)
{
	if (str.empty()) return;
	strreplace(str, "&", "&amp;");
	strreplace(str, "<", "&lt;");
	strreplace(str, ">", "&gt;");
}

void ChatProcessor::sendNodes(const TCPSocket& s, vector<NodeInfo>& nodes)
{
	string nodesxml = "<o2on>";
	for (vector<NodeInfo>::iterator nodesit = nodes.begin();
		nodesit != nodes.end();
		++nodesit)
	{
		nodesxml += boost::str(boost::format("<node host=\"%1%\" port=\"%2%\"/>")
			% nodesit->getHost() % nodesit->getPort());
	}
	nodesxml += "</o2on>";

	sendP2PMessage(s, "NODE", vector<char>(nodesxml.begin(), nodesxml.end()));
}

void ChatProcessor::acceptDetected()
{
	setChatMessage("System", "Inbound connection.");
	updateStat(*server_);
}

bool ChatProcessor::recvDetected(TCPSocketBuffer<TimedTCPSocket>& sockbuf)
{
	if (sockbuf.getBuffer().length() > 1024 * 1024) {
		setChatMessage("System", "Kill Connection. (too long data)");
		return false;
	}
	//if (strfound(sockbuf.getBuffer().find("</o2on>"))) {
	//	// ここでxmlをパースしてパースした部分をbufferから取り除く
	//	if (!parseProtocol(sockbuf)) {
	//		// パースに失敗したので切断
	//		setChatMessage("System", "Kill connection. (Invalid messages)");
	//		return false;
	//	}
	//}
	string& str = sockbuf.getBuffer();

	size_t pos;
	while (pos = str.find("\r\n\r\n"), strfound(pos)) {
		string header(str, 0, pos);
		istringstream iss(header);
		int length = 0;
		string method;
		while (iss) {
			string line;
			if (!getline(iss, line)) break;
			size_t pos2 = line.find(": ");
			if (!strfound(pos2)) continue;
			string key(line, 0, pos2);

			string value(line, pos2 + 2);
			if (strfound(pos2 = value.find("\r"))) {
				value.erase(pos2);
			}

			if (key == "Method") {
				method = value;
			} else if (key == "Content-Length") {
				try {
					length = boost::lexical_cast<int>(value);
				} catch (boost::bad_lexical_cast&) {}
			}
		}
		if (str.size() < header.size() + 4 + length) return true;	// まだ全部届いてない

		str.erase(0, pos + 4); // Headerをバッファから消す

		if (length == 0) return true;

		vector<char> deflateBody(str.begin(), str.begin() + length);
		str.erase(0, length);
		vector<char> body;
		if (!zlib_inflate(deflateBody, body)) {
			// 解凍失敗
			setChatMessage("System", "Kill connection. (deflate failed)");
			return false;
		}

		string strbody(body.begin(), body.end());
		//string strbody(deflateBody.begin(), deflateBody.end());
		//string tmp = "[" + method + "]:[" + strbody + "]";
		//MessageBox(NULL, tmp.c_str(), "hogehoge - o2on", MB_OK);
		auto_ptr<XMLParser> parser;
		if (method == "MESSAGE") {
			parser.reset(new MessageXMLParser(*this));
		} else if (method == "NODE") {
			parser.reset(new NodeXMLParser(this));
		} else if (method == "HANDSHAKE") {
			parser.reset(new HandshakeXMLParser(*this, sockbuf));
		}

		if (parser.get()) {
			if (!parser->parse(strbody))
				return false;

			updateStat(*server_);
		}
	}

	return true;
}

void ChatProcessor::disconnected()
{
	setChatMessage("System", "Connection killed.");
	updateStat(*server_);
}

void ChatProcessor::dosomething(list<TCPSocketBuffer<TimedTCPSocket> >& sockets)
{
	// ChatMessageを周りのノードに投げる
	if (!msgqueue.empty()) {
		Lock lock(cs_msgq);
		while (!msgqueue.empty()) {
			for (list<TCPSocketBuffer<TimedTCPSocket> >::iterator it = sockets.begin(),
				end = sockets.end();
				it != end;
				++it)
			{
				const string buf(msgqueue.front().toXmlString());
				sendP2PMessage(it->getSocket(), "MESSAGE", vector<char>(buf.begin(), buf.end()));
			}
			msgqueue.pop();
		}
	}

	time_t nowtime;
	time(&nowtime);
	if (nowtime == pretime) {
		// 1sec経っていない場合は何もしない
		return;
	} else {
		pretime = nowtime;
	}

	// 90秒以上何も受信していない物を切断
	for (list<TCPSocketBuffer<TimedTCPSocket> >::iterator it = sockets.begin();
		it != sockets.end();
		)
	{
		if (it->getSocket().recvTimeout(90)) {
			setChatMessage("System", "Timed out. (90sec)");
			it = server_->pop(it);
			disconnected();
		} else {
			++it;
		}
	}

	// 60秒以上何も送信していない物にnodeを撃つ
	for (list<TCPSocketBuffer<TimedTCPSocket> >::iterator it(sockets.begin()), end(sockets.end());
		it != end;
		++it)
	{
		if (it->getSocket().sendTimeout(60)) {
			//const static string noop("<?xml version=\"1.0\" encoding=\"Shift_JIS\"?>"
			//		"<o2on service=\"system\" type=\"noop\"></o2on>");
			//it->send(noop.data(), noop.length());
			sendNodes(it->getSocket(), nodes);
		}
	}

	using namespace boost::lambda;

	// 接続数が少ない場合外部に接続
	if (sockets.size() < maxOutConnection && nodes.size() > sockets.size()
		&& waitingSockets.size() < maxWaitingConnection) {
		// 優先度が高いほうから sockets.size() + 1 個のノードを取得
		vector<NodeInfo>::iterator nth = nodes.begin() + sockets.size() + 1;
		nth_element(nodes.begin(), nth, nodes.end(),
			(bind(&NodeInfo::getPriority, _1) > bind(&NodeInfo::getPriority, _2)));

		for (vector<NodeInfo>::iterator it = nodes.begin(); it != nth; ++it) {
			if (find_if(sockets.begin(), sockets.end(), findFromAddr(it->getAddr()))
				!= sockets.end())
			{
				continue;
			}

            TCPSocket new_sock;
			new_sock.nonblocking();
			if (new_sock.connect(it->getAddr(), it->getPort())) {
				SOCKET sock = new_sock.release();
#pragma warning(push, 3)
				FD_SET(sock, &waitingFd);
#pragma warning(pop)
				waitingSockets.push_back(sock);
			}
			it->decPriority();
			break;
		}
	}

	fd_set waiting = waitingFd;
	timeval waitval = {0, 1};
	// 接続できた物とハンドシェイクしてserventに登録
	if (select(FD_SETSIZE, NULL, &waiting, NULL, &waitval) > 0) {
		for (vector<SOCKET>::iterator it = waitingSockets.begin();
			it != waitingSockets.end();
			)
		{
			if (FD_ISSET(*it, &waiting)) {
				TCPSocket new_sock(*it);
				sockaddr_in sin;
				new_sock.getpeername(reinterpret_cast<sockaddr*>(&sin));
				vector<NodeInfo>::iterator nodesit = find_if(nodes.begin(), nodes.end(),
					bind(&NodeInfo::getAddr, _1) == sin.sin_addr.s_addr);
				if (nodesit != nodes.end()) {
					nodesit->incPriority();
				}

				outboundHandshake(new_sock, chat_port_, sin.sin_addr.s_addr);
				sendNodes(new_sock, nodes);	// 暫定的にここでノードを投げる
				server_->push(new_sock.release());
				setChatMessage("System", "Connected to outbound.");
				updateStat(*server_);

#pragma warning(push, 3)
				FD_CLR(*it, &waitingFd);
#pragma warning(pop)
				it = waitingSockets.erase(it);
			} else {
				++it;
			}
		}
	}

	// 接続に失敗した物を除去
	waiting = waitingFd;
	if (select(FD_SETSIZE, NULL, NULL, &waiting, &waitval) > 0) {
		for (vector<SOCKET>::iterator it = waitingSockets.begin();
			it != waitingSockets.end();
			)
		{
			if (FD_ISSET(*it, &waiting)) {
				shutdown(*it, 1);
				closesocket(*it);
#pragma warning(push, 3)
				FD_CLR(*it, &waitingFd);
#pragma warning(pop)
				it = waitingSockets.erase(it);
			} else {
				++it;
			}
		}
	}
}

void ChatProcessor::startProcessor()
{
	//// MSXML初期化
	//CoInitialize(NULL);
	//HRESULT hr = pDoc.CreateInstance(MSXML2::CLSID_DOMDocument);
	//if (FAILED(hr)) {
	//	throw exception("MSXML3初期化失敗\nChat service は開始されませんでした");
	//}
	//pDoc->put_async(VARIANT_FALSE);
	//pDoc->setProperty(L"SelectionLanguage", _variant_t(L"XPath"));

	// Node読み込み
	ifstream nodeifs(chatNodeFileName);
	if (!nodeifs.fail()) {
		string buf;
		while (getline(nodeifs, buf), !nodeifs.fail()) {
			try {
				size_t pos = buf.rfind(":");
				pushNode(NodeInfo(buf.substr(0, pos),
					boost::lexical_cast<unsigned short>(buf.substr(pos + 1))));
			} catch (exception&) {}
		}
	}
}

ChatProcessor::~ChatProcessor()
{
	// Node書き出し
	ofstream nodeofs(chatNodeFileName);
	if (!nodeofs.fail()) {
		for (vector<NodeInfo>::iterator it = nodes.begin(), end = nodes.end();
			it != end;
			++it)
		{
			nodeofs << it->getHost() << ':' << boost::lexical_cast<string>(it->getPort()) << '\n';
		}
	}

	for (vector<SOCKET>::iterator it(waitingSockets.begin()), end(waitingSockets.end());
		it != end;
		++it)
	{
		shutdown(*it, 1);
		closesocket(*it);
	}

	//pDoc.Release();
	//CoUninitialize();
}

string ipaddr2string(unsigned long addr)
{
	return boost::str(boost::format("%d.%d.%d.%d") %
		(addr & 0xff) % ((addr >> 8) & 0xff) % ((addr >> 16) & 0xff) % ((addr >> 24) & 0xff));
}

void ChatProcessor::pushNode(const NodeInfo& node)
{
	using namespace boost::lambda;
	Lock lock(cs_nodes);

	if (find(nodes.begin(), nodes.end(), node) != nodes.end()) return;

	nodes.push_back(node);
	
	if (nodes.size() > 100) { // 100以上の時は優先度が低いほうから削り80まで減らす
		vector<NodeInfo>::iterator nth = nodes.begin() + (nodes.size() - 80);
		nth_element(nodes.begin(), nth, nodes.end(),
			(bind(&NodeInfo::getPriority, _1) < bind(&NodeInfo::getPriority, _2)));
		nodes.erase(nodes.begin(), nth);
		vector<NodeInfo>(nodes).swap(nodes);
	}
}

void XMLCALL NodeXMLParser::nodeElementStart(void* data, const char* el, const char** attr)
{
	ChatProcessor* chatProcessor = static_cast<ChatProcessor*>(data);

	string element = el;
	if (element == "node") {
		string host;
		unsigned short port = 0;

		for (int i = 0; attr[i]; i += 2) {
			string name = attr[i];
			if (name == "host") {
				host = attr[i+1];
			} else if (name == "port") {
				try {
					port = boost::lexical_cast<unsigned short>(attr[i+1]);
				} catch (boost::bad_lexical_cast&) {}
			}
		}

		if (!host.empty() && port != 0) {
			try {
				chatProcessor->pushNode(NodeInfo(host, port));
			} catch (exception&) {}
		}
	}
}

void XMLCALL HandshakeXMLParser::elementStart(void* data, const char* el, const char** attr)
{
	HandshakeXMLParser* parser = static_cast<HandshakeXMLParser*>(data);

	string element = el;
	if (element == "nodeinfo") {
		unsigned short port = 0;

		for (int i = 0; attr[i]; i += 2) {
			string name = attr[i];
			if (name == "myport") {
				try {
					port = boost::lexical_cast<unsigned short>(attr[i+1]);
				} catch (boost::bad_lexical_cast&) {}
			}
		}

		if (port != 0) {
			sockaddr_in sin;
			parser->sockbuf_.getpeername(reinterpret_cast<sockaddr *>(&sin));
			string ipaddr = ipaddr2string(sin.sin_addr.s_addr);
			try {
				parser->chatProcessor_.pushNode(NodeInfo(ipaddr, port));
			} catch (exception&) {}
		}
	}
}

void XMLCALL MessageXMLParser::elementStart(void* data, const char* el, const char** /*attr*/)
{
	MessageXMLParser* parser = static_cast<MessageXMLParser*>(data);

	parser->element = el;
}

void XMLCALL MessageXMLParser::charHandler(void* data, const XML_Char* s, int len)
{
	MessageXMLParser* parser = static_cast<MessageXMLParser*>(data);

	if (parser->element == "guid") {
		parser->guid.append(s, len);
	} else if (parser->element == "name") {
		parser->name.append(s, len);
	} else if (parser->element == "message") {
		parser->message.append(s, len);
	} else if (parser->element == "ttl") {
		try {
			parser->ttl = lexical_cast<int>(string(s, len));
		} catch (boost::bad_lexical_cast&) {}
	} else if (parser->element == "id") {
		parser->id.append(s, len);
	} else if (parser->element == "sign") {
		parser->sign.append(s, len);
	}
}

void XMLCALL MessageXMLParser::elementEnd(void* data, const char* el)
{
	string element = el;
	if (element != "o2on")
		return;

	MessageXMLParser* parser = static_cast<MessageXMLParser*>(data);
	ChatProcessor& cp = parser->chatProcessor_;

	Lock lock(cp.getMsgqCS());
	deque<string>& processed = cp.getProcessed();
	if (find(processed.begin(), processed.end(), parser->guid) == processed.end()) {
		// このGUIDを持つメッセージは始めて到達した
		processed.push_back(parser->guid);
		if (processed.size() > 100) {
			processed.pop_front();
		}
		time_t msgtime;
		time(&msgtime);
		setChatMessage(parser->name, parser->message, msgtime);

		Lock lock2(cp.getMsgCS());
		--(parser->ttl);
		ChatMessage chatmsg(parser->name, parser->message, parser->guid, msgtime, parser->id,
			parser->sign, parser->ttl);
		cp.getMessages().push_back(chatmsg);
		if (cp.getMessages().size() > 500) {
			cp.getMessages().pop_front();
		}

		// TTLが正なら周りにもう一回投げる
		if (parser->ttl > 0) {
			cp.getMsgqueue().push(chatmsg);
		}
	}
}
