[网络协议]KBEngine奇技淫巧＜五＞

网络模块的组织

KBEngine最核心的三个模块我梳理的还剩下网络，这篇文章就来聊聊。网络主要是两个模块，一个是EventPoller，一个是NetworkInterface。

EventPoller

EventPoller是一个网络模块借口的抽象，由于是服务器编程，就需要涉及到网络模式，我们常说的windows里面的网络模式，select，select事件，重叠IO，和完成端口。Linux主要是select，poll和epoll。KBEngine分不同平台不同的实现。在调试开发阶段，标哥提倡用windows平台。windows平台的网络实现是select模式，linux的网络实现是epoll。这样就实现的管理各种socket的任务。

结构图

很明显，网络模块要比Task和timer复杂的多，虽然里面设计理念还有注册等待回调的影子，但是复杂度提升了一个级别。

对象组织

我们还是来看看每一个类的实例是怎么创建来的。在EventDispatcher里面有一个EventPoller变量，我们可以看到。
Eventpoller是根据平台来实现的，由C++编译宏控制，这个宏就是：

#if KBE_PLATFORM != PLATFORM_WIN32
#define HAS_EPOLL
#endif

在EventDispathcer里面调用创建poller，在函数里面根据不同平台创建对应平台的poller。

EventPoller* EventDispatcher::createPoller()
{
	pPoller_ = EventPoller::create();
	return pPoller_;
}

EventPoller * EventPoller::create()
{
#ifdef HAS_EPOLL
	return new EpollPoller();
#else
	return new SelectPoller();
#endif // HAS_EPOLL
}

搜索一番好像没有调用点调用create，不过在构造函数里面调用创建：

EventDispatcher::EventDispatcher() :
	breakProcessing_(EVENT_DISPATCHER_STATUS_RUNNING),
	maxWait_(0.1),
	numTimerCalls_(0),
	accSpareTime_(0),
	oldSpareTime_(0),
	totSpareTime_(0),
	lastStatisticsGathered_(0),
	pTasks_(new Tasks),
	pErrorReporter_(NULL),
	pTimers_(new Timers64)
	
{
	pPoller_ = EventPoller::create();
	pErrorReporter_ = new ErrorReporter(*this);
}

这也能看出来标哥在写的时候有时候就是设计了，回头调用还是原生代码。我们可以找到poller最核心的两个变量：
这两个变量都是容器，容器类型是：

class InputNotificationHandler;
typedef std::map<int, InputNotificationHandler *> FDReadHandlers;
typedef std::map<int, OutputNotificationHandler *> FDWriteHandlers;

最主要的功能就是利用inputnotificationhandler和outputnotificationhandler来实现的，代表输入和输出操作。

InputNotificationHandler

这是两个基类，在输入操作里面分为：
主要是listener和receive，receive又分为UDP和TCP，UDP标哥又重新封装了一下称为KCP。

如何创建各种Handler

我们知道socket编程里面，服务器端里面需要绑定监听，然后建立的连接是新的socket，虽然有封装但是离不开原理。
在EventDispatcher里面封装了两个函数是注册输入和注册输出：

bool EventDispatcher::registerReadFileDescriptor(int fd,
	InputNotificationHandler * handler)
{
	return pPoller_->registerForRead(fd, handler);
}

//-------------------------------------------------------------------------------------
bool EventDispatcher::registerWriteFileDescriptor(int fd,
	OutputNotificationHandler * handler)
{
	return pPoller_->registerForWrite(fd, handler);
}

网络接口NetworkInterface在初始化的时候会创建listener，注册到poller里面，也就是实现了绑定和监听端口：

bool NetworkInterface::initialize(const char* pEndPointName, uint16 listeningPort_min, uint16 listeningPort_max,
										const char * listeningInterface, EndPoint* pEP, ListenerReceiver* pLR, uint32 rbuffer, 
										uint32 wbuffer)
{
	KBE_ASSERT(listeningInterface && pEP && pLR);

	if (pEP->good())
	{
		this->dispatcher().deregisterReadFileDescriptor(*pEP);
		pEP->close();
	}

	Address address;
	address.ip = 0;
	address.port = 0;

	bool isTCP = strstr(pEndPointName, "-TCP") != NULL;

	if(isTCP)
		pEP->socket(SOCK_STREAM);
	else
		pEP->socket(SOCK_DGRAM);

	if (!pEP->good())
	{
		ERROR_MSG(fmt::format("NetworkInterface::initialize({}): couldn't create a socket\n",
			pEndPointName));

		return false;
	}
	
	if (listeningPort_min > 0 && listeningPort_min == listeningPort_max)
		pEP->setreuseaddr(true);
	
	this->dispatcher().registerReadFileDescriptor(*pEP, pLR);
	
	u_int32_t ifIPAddr = INADDR_ANY;

	bool listeningInterfaceEmpty =
		(listeningInterface == NULL || listeningInterface[0] == 0);

	// 查找指定接口名 NIP、MAC、IP是否可用
	if(pEP->findIndicatedInterface(listeningInterface, ifIPAddr) == 0)
	{
		char szIp[MAX_IP] = {0};
		Address::ip2string(ifIPAddr, szIp);

		INFO_MSG(fmt::format("NetworkInterface::initialize({}): Creating on interface '{}' (= {})\n",
			pEndPointName, listeningInterface, szIp));
	}

	// 如果不为空又找不到那么警告用户错误的设置，同时我们采用默认的方式(绑定到INADDR_ANY)
	else if (!listeningInterfaceEmpty)
	{
		WARNING_MSG(fmt::format("NetworkInterface::initialize({}): Couldn't parse interface spec '{}' so using all interfaces\n",
			pEndPointName, listeningInterface));
	}
	
	// 尝试绑定到端口，如果被占用向后递增
	bool foundport = false;
	uint32 listeningPort = listeningPort_min;
	if(listeningPort_min != listeningPort_max)
	{
		for(int lpIdx=ntohs(listeningPort_min); lpIdx<=ntohs(listeningPort_max); ++lpIdx)
		{
			listeningPort = htons(lpIdx);
			if (pEP->bind(listeningPort, ifIPAddr) != 0)
			{
				continue;
			}
			else
			{
				foundport = true;
				break;
			}
		}
	}
	else
	{
		if (pEP->bind(listeningPort, ifIPAddr) == 0)
		{
			foundport = true;
		}
	}

	// 如果无法绑定到合适的端口那么报错返回，进程将退出
	if(!foundport)
	{
		ERROR_MSG(fmt::format("NetworkInterface::initialize({}): Couldn't bind the socket to {}:{} ({})\n",
			pEndPointName, inet_ntoa((struct in_addr&)ifIPAddr), ntohs(listeningPort), kbe_strerror()));
		
		pEP->close();
		return false;
	}

	// 获得当前绑定的地址，如果是INADDR_ANY这里获得的IP是0
	pEP->getlocaladdress( (u_int16_t*)&address.port,
		(u_int32_t*)&address.ip );

	if (0 == address.ip)
	{
		u_int32_t addr;
		if(0 == pEP->getDefaultInterfaceAddress(addr))
		{
			address.ip = addr;

			char szIp[MAX_IP] = {0};
			Address::ip2string(address.ip, szIp);
			INFO_MSG(fmt::format("NetworkInterface::initialize({}): bound to all interfaces with default route interface on {} ( {} )\n",
				pEndPointName, szIp, address.c_str()));
		}
		else
		{
			ERROR_MSG(fmt::format("NetworkInterface::initialize({}): Couldn't determine ip addr of default interface\n", pEndPointName));

			pEP->close();
			return false;
		}
	}
	
	pEP->setnonblocking(true);
	pEP->setnodelay(true);
	pEP->addr(address);
	
	if(rbuffer > 0)
	{
		if (!pEP->setBufferSize(SO_RCVBUF, rbuffer))
		{
			WARNING_MSG(fmt::format("NetworkInterface::initialize({}): Operating with a receive buffer of only {} bytes (instead of {})\n",
				pEndPointName, pEP->getBufferSize(SO_RCVBUF), rbuffer));
		}
	}
	if(wbuffer > 0)
	{
		if (!pEP->setBufferSize(SO_SNDBUF, wbuffer))
		{
			WARNING_MSG(fmt::format("NetworkInterface::initialize({}): Operating with a send buffer of only {} bytes (instead of {})\n",
				pEndPointName, pEP->getBufferSize(SO_SNDBUF), wbuffer));
		}
	}


	int backlog = Network::g_SOMAXCONN;
	if (backlog < 5)
		backlog = 5;

	if (isTCP)
	{
		if (pEP->listen(backlog) == -1)
		{
			ERROR_MSG(fmt::format("NetworkInterface::initialize({}): listen to {} ({})\n",
				pEndPointName, address.c_str(), kbe_strerror()));

			pEP->close();
			return false;
		}
	}

	INFO_MSG(fmt::format("NetworkInterface::initialize({}): address {}, SOMAXCONN={}.\n", 
		pEndPointName, address.c_str(), backlog));

	return true;
}

LIstener初始化

网路接口是app的成员，在app初始化的时候会调用网络接口的初始化：
NetworkInterface在构造函数里面直接绑定端口：

NetworkInterface::NetworkInterface(Network::EventDispatcher * pDispatcher,
		int32 extlisteningTcpPort_min, int32 extlisteningTcpPort_max, int32 extlisteningUdpPort_min, int32 extlisteningUdpPort_max, const char * extlisteningInterface,
		uint32 extrbuffer, uint32 extwbuffer,
		int32 intlisteningPort_min, int32 intlisteningPort_max, const char * intlisteningInterface,
		uint32 intrbuffer, uint32 intwbuffer):
	extTcpEndpoint_(),
	extUdpEndpoint_(),
	intTcpEndpoint_(),
	channelMap_(),
	pDispatcher_(pDispatcher),
	pExtListenerReceiver_(NULL),
	pExtUdpListenerReceiver_(NULL),
	pIntListenerReceiver_(NULL),
	pDelayedChannels_(new DelayedChannels()),
	pChannelTimeOutHandler_(NULL),
	pChannelDeregisterHandler_(NULL),
	numExtChannels_(0)
{
	if(extlisteningTcpPort_min != -1)
	{
		pExtListenerReceiver_ = new ListenerTcpReceiver(extTcpEndpoint_, Channel::EXTERNAL, *this);

		this->initialize("EXTERNAL-TCP", htons(extlisteningTcpPort_min), htons(extlisteningTcpPort_max),
			extlisteningInterface, &extTcpEndpoint_, pExtListenerReceiver_, extrbuffer, extwbuffer);

		// 如果配置了对外端口范围， 如果范围过小这里extEndpoint_可能没有端口可用了
		if(extlisteningTcpPort_min != -1)
		{
			KBE_ASSERT(extTcpEndpoint_.good() && "Channel::EXTERNAL-TCP: no available port, "
				"please check for kbengine[_defs].xml!\n");
		}
	}

	if (extlisteningUdpPort_min != -1)
	{
		pExtUdpListenerReceiver_ = new ListenerUdpReceiver(extUdpEndpoint_, Channel::EXTERNAL, *this);

		this->initialize("EXTERNAL-UDP", htons(extlisteningUdpPort_min), htons(extlisteningUdpPort_max),
			extlisteningInterface, &extUdpEndpoint_, pExtUdpListenerReceiver_, extrbuffer, extwbuffer);

		// 如果配置了对外端口范围， 如果范围过小这里extEndpoint_可能没有端口可用了
		if (extlisteningUdpPort_min != -1)
		{
			KBE_ASSERT(extUdpEndpoint_.good() && "Channel::EXTERNAL-UDP: no available udp-port, "
				"please check for kbengine[_defs].xml!\n");
		}
	}

	if (intlisteningPort_min != -1)
	{
		pIntListenerReceiver_ = new ListenerTcpReceiver(intTcpEndpoint_, Channel::INTERNAL, *this);

		this->initialize("INTERNAL-TCP", htons(intlisteningPort_min), htons(intlisteningPort_max),
			intlisteningInterface, &intTcpEndpoint_, pIntListenerReceiver_, intrbuffer, intwbuffer);
	}

	KBE_ASSERT(good() && "NetworkInterface::NetworkInterface: no available port, "
		"please check for kbengine[_defs].xml!\n");

	pDelayedChannels_->init(this->dispatcher(), this);
}

Receiver创建

创建了Listener以后，receiver就是监听的时候建立连接，注册到poller里面。

int ListenerTcpReceiver::handleInputNotification(int fd)
{
	int tickcount = 0;

	while(tickcount ++ < 256)
	{
		EndPoint* pNewEndPoint = endpoint_.accept();
		if(pNewEndPoint == NULL){

			if(tickcount == 1)
			{
				WARNING_MSG(fmt::format("ListenerTcpReceiver::handleInputNotification: accept endpoint({}) {}! channelSize={}\n",
					fd, kbe_strerror(), networkInterface_.channels().size()));
				
				this->dispatcher().errorReporter().reportException(
						REASON_GENERAL_NETWORK);
			}

			break;
		}
		else
		{
			Channel* pChannel = Network::Channel::createPoolObject(OBJECTPOOL_POINT);
			bool ret = pChannel->initialize(networkInterface_, pNewEndPoint, traits_);
			if(!ret)
			{
				ERROR_MSG(fmt::format("ListenerTcpReceiver::handleInputNotification: initialize({}) is failed!\n",
					pChannel->c_str()));

				pChannel->destroy();
				Network::Channel::reclaimPoolObject(pChannel);
				return 0;
			}

			if(!networkInterface_.registerChannel(pChannel))
			{
				ERROR_MSG(fmt::format("ListenerTcpReceiver::handleInputNotification: registerChannel({}) is failed!\n",
					pChannel->c_str()));

				pChannel->destroy();
				Network::Channel::reclaimPoolObject(pChannel);
			}
		}
	}

	return 0;
}

Channel

一个channel就是一个socket连接，在channel里面有成员变量receiver和sender
初始化的时候会注册到poller里面

bool Channel::initialize(NetworkInterface & networkInterface, 
		const EndPoint * pEndPoint, 
		Traits traits, 
		ProtocolType pt,
		ProtocolSubType spt,
		PacketFilterPtr pFilter, 
		ChannelID id)
{
	id_ = id;
	protocoltype_ = pt;
	protocolSubtype_ = spt;
	traits_ = traits;
	pFilter_ = pFilter;
	pNetworkInterface_ = &networkInterface;
	this->pEndPoint(pEndPoint);

	KBE_ASSERT(pNetworkInterface_ != NULL);
	KBE_ASSERT(pEndPoint_ != NULL);

	if(protocoltype_ == PROTOCOL_TCP)
	{
		if(pPacketReceiver_)
		{
			if(pPacketReceiver_->type() == PacketReceiver::UDP_PACKET_RECEIVER)
			{
				SAFE_RELEASE(pPacketReceiver_);
				pPacketReceiver_ = new TCPPacketReceiver(*pEndPoint_, *pNetworkInterface_);
			}
		}
		else
		{
			pPacketReceiver_ = new TCPPacketReceiver(*pEndPoint_, *pNetworkInterface_);
		}

		KBE_ASSERT(pPacketReceiver_->type() == PacketReceiver::TCP_PACKET_RECEIVER);

		// UDP不需要注册描述符
		pNetworkInterface_->dispatcher().registerReadFileDescriptor(*pEndPoint_, pPacketReceiver_);

		// 需要发送数据时再注册
		// pPacketSender_ = new TCPPacketSender(*pEndPoint_, *pNetworkInterface_);
		// pNetworkInterface_->dispatcher().registerWriteFileDescriptor(*pEndPoint_, pPacketSender_);

		if (pPacketSender_ && pPacketSender_->type() != PacketSender::TCP_PACKET_SENDER)
		{
			KCPPacketSender::reclaimPoolObject((KCPPacketSender*)pPacketSender_);
			pPacketSender_ = NULL;
		}
	}
	else
	{
		if (protocolSubtype_ == SUB_PROTOCOL_KCP)
		{
			if (pPacketReceiver_)
			{
				if (pPacketReceiver_->type() == PacketReceiver::TCP_PACKET_RECEIVER)
				{
					SAFE_RELEASE(pPacketReceiver_);
					pPacketReceiver_ = new KCPPacketReceiver(*pEndPoint_, *pNetworkInterface_);
				}
			}
			else
			{
				pPacketReceiver_ = new KCPPacketReceiver(*pEndPoint_, *pNetworkInterface_);
			}

			if (!init_kcp())
			{
				KBE_ASSERT(false);
				return false;
			}
		}
		else
		{
			if (pPacketReceiver_)
			{
				if (pPacketReceiver_->type() == PacketReceiver::TCP_PACKET_RECEIVER)
				{
					SAFE_RELEASE(pPacketReceiver_);
					pPacketReceiver_ = new UDPPacketReceiver(*pEndPoint_, *pNetworkInterface_);
				}
			}
			else
			{
				pPacketReceiver_ = new UDPPacketReceiver(*pEndPoint_, *pNetworkInterface_);
			}
		}

		KBE_ASSERT(pPacketReceiver_->type() == PacketReceiver::UDP_PACKET_RECEIVER);

		if (pPacketSender_ && pPacketSender_->type() != PacketSender::UDP_PACKET_SENDER)
		{
			TCPPacketSender::reclaimPoolObject((TCPPacketSender*)pPacketSender_);
			pPacketSender_ = NULL;
		}
	}

	pPacketReceiver_->pEndPoint(pEndPoint_);
	
	if(pPacketSender_)
		pPacketSender_->pEndPoint(pEndPoint_);

	startInactivityDetection((traits_ == INTERNAL) ? g_channelInternalTimeout : 
													g_channelExternalTimeout,
							(traits_ == INTERNAL) ? g_channelInternalTimeout / 2.f: 
													g_channelExternalTimeout / 2.f);

	return true;
}

注册到Poller里面以后就可以收发协议了。里面有个Reader变量管理着如何解析协议

	PacketReader*				pPacketReader_;

各种协议Handler的调用

这个东西就会调用我们最开始，原先文章说的协议handler，

void PacketReader::processMessages(KBEngine::Network::MessageHandlers* pMsgHandlers, Packet* pPacket)
{
	while(pPacket->length() > 0 || pFragmentStream_ != NULL)
	{
		if(fragmentDatasFlag_ == FRAGMENT_DATA_UNKNOW)
		{
			// 如果没有ID信息，先获取ID
			if(currMsgID_ == 0)
			{
				if(NETWORK_MESSAGE_ID_SIZE > 1 && pPacket->length() < NETWORK_MESSAGE_ID_SIZE)
				{
					writeFragmentMessage(FRAGMENT_DATA_MESSAGE_ID, pPacket, NETWORK_MESSAGE_ID_SIZE);
					break;
				}

				(*pPacket) >> currMsgID_;
				pPacket->messageID(currMsgID_);
			}

			Network::MessageHandler* pMsgHandler = pMsgHandlers->find(currMsgID_);

			if(pMsgHandler == NULL)
			{
				MemoryStream* pPacket1 = pFragmentStream_ != NULL ? pFragmentStream_ : pPacket;
				TRACE_MESSAGE_PACKET(true, pPacket1, pMsgHandler, pPacket1->length(), pChannel_->c_str(), false);
				
				// 用作调试时比对
				uint32 rpos = pPacket1->rpos();
				pPacket1->rpos(0);
				TRACE_MESSAGE_PACKET(true, pPacket1, pMsgHandler, pPacket1->length(), pChannel_->c_str(), false);
				pPacket1->rpos(rpos);

				ERROR_MSG(fmt::format("PacketReader::processMessages: not found msgID={}, msglen={}, from {}.\n",
					currMsgID_, pPacket1->length(), pChannel_->c_str()));

				currMsgID_ = 0;
				currMsgLen_ = 0;
				pChannel_->condemn("PacketReader::processMessages: not found msgID");
				break;
			}

			// 如果没有可操作的数据了则退出等待下一个包处理。
			// 可能是一个无参数数据包
			//if(pPacket->opsize() == 0)	
			//	break;
			
			// 如果长度信息没有获得，则等待获取长度信息
			if(currMsgLen_ == 0)
			{
				// 如果长度信息是可变的或者配置了永远包含长度信息选项时，从流中分析长度数据
				if(pMsgHandler->msgLen == NETWORK_VARIABLE_MESSAGE)
				{
					// 如果长度信息不完整，则等待下一个包处理
					if(pPacket->length() < NETWORK_MESSAGE_LENGTH_SIZE)
					{
						writeFragmentMessage(FRAGMENT_DATA_MESSAGE_LENGTH, pPacket, NETWORK_MESSAGE_LENGTH_SIZE);
						break;
					}
					else
					{
						// 此处获得了长度信息
						Network::MessageLength currlen;
						(*pPacket) >> currlen;
						currMsgLen_ = currlen;

						NetworkStats::getSingleton().trackMessage(NetworkStats::RECV, *pMsgHandler, 
							currMsgLen_ + NETWORK_MESSAGE_ID_SIZE + NETWORK_MESSAGE_LENGTH_SIZE);

						if (currMsgLen_ == NETWORK_MESSAGE_MAX_SIZE)
							currMsgLen_ = NETWORK_MESSAGE_MAX_SIZE1;
					}
				}
				else
				{
					currMsgLen_ = pMsgHandler->msgLen;

					NetworkStats::getSingleton().trackMessage(NetworkStats::RECV, *pMsgHandler, 
						currMsgLen_ + NETWORK_MESSAGE_LENGTH_SIZE);
				}
			}

			// 如果长度占满说明使用了扩展长度，我们还需要等待扩展长度信息
			if (currMsgLen_ == NETWORK_MESSAGE_MAX_SIZE1)
			{
				if (pPacket->length() < NETWORK_MESSAGE_LENGTH1_SIZE)
				{
					// 如果长度信息不完整，则等待下一个包处理
					writeFragmentMessage(FRAGMENT_DATA_MESSAGE_LENGTH1, pPacket, NETWORK_MESSAGE_LENGTH1_SIZE);
					break;
				}
				else
				{
					// 此处获得了扩展长度信息
					(*pPacket) >> currMsgLen_;

					NetworkStats::getSingleton().trackMessage(NetworkStats::RECV, *pMsgHandler,
						currMsgLen_ + NETWORK_MESSAGE_ID_SIZE + NETWORK_MESSAGE_LENGTH1_SIZE);
				}
			}

			if(this->pChannel_->isExternal() && 
				g_componentType != BOTS_TYPE && 
				g_componentType != CLIENT_TYPE && 
				currMsgLen_ > NETWORK_MESSAGE_MAX_SIZE)
			{
				MemoryStream* pPacket1 = pFragmentStream_ != NULL ? pFragmentStream_ : pPacket;
				TRACE_MESSAGE_PACKET(true, pPacket1, pMsgHandler, pPacket1->length(), pChannel_->c_str(), false);

				// 用作调试时比对
				uint32 rpos = pPacket1->rpos();
				pPacket1->rpos(0);
				TRACE_MESSAGE_PACKET(true, pPacket1, pMsgHandler, pPacket1->length(), pChannel_->c_str(), false);
				pPacket1->rpos(rpos);

				WARNING_MSG(fmt::format("PacketReader::processMessages({0}): msglen exceeds the limit! msgID={1}, msglen=({2}:{3}), maxlen={5}, from {4}.\n", 
					pMsgHandler->name.c_str(), currMsgID_, currMsgLen_, pPacket1->length(), pChannel_->c_str(), NETWORK_MESSAGE_MAX_SIZE));

				currMsgLen_ = 0;
				pChannel_->condemn("PacketReader::processMessages: msglen exceeds the limit!");
				break;
			}

			if(pFragmentStream_ != NULL)
			{
				TRACE_MESSAGE_PACKET(true, pFragmentStream_, pMsgHandler, currMsgLen_, pChannel_->c_str(), false);
				pMsgHandler->handle(pChannel_, *pFragmentStream_);
				MemoryStream::reclaimPoolObject(pFragmentStream_);
				pFragmentStream_ = NULL;
			}
			else
			{
				if(pPacket->length() < currMsgLen_)
				{
					writeFragmentMessage(FRAGMENT_DATA_MESSAGE_BODY, pPacket, currMsgLen_);
					break;
				}

				// 临时设置有效读取位， 防止接口中溢出操作
				size_t wpos = pPacket->wpos();
				// size_t rpos = pPacket->rpos();
				size_t frpos = pPacket->rpos() + currMsgLen_;
				pPacket->wpos(frpos);

				TRACE_MESSAGE_PACKET(true, pPacket, pMsgHandler, currMsgLen_, pChannel_->c_str(), true);
				pMsgHandler->handle(pChannel_, *pPacket);

				// 如果handler没有处理完数据则输出一个警告
				if(currMsgLen_ > 0)
				{
					if(frpos != pPacket->rpos())
					{
						WARNING_MSG(fmt::format("PacketReader::processMessages({}): rpos({}) invalid, expect={}. msgID={}, msglen={}.\n",
							pMsgHandler->name.c_str(), pPacket->rpos(), frpos, currMsgID_, currMsgLen_));

						pPacket->rpos(frpos);
					}
				}

				pPacket->wpos(wpos);
			}

			currMsgID_ = 0;
			currMsgLen_ = 0;
		}
		else
		{
			mergeFragmentMessage(pPacket);
		}
	}
}

转了一个大圈终于回调到了具体的业务。

NetworkInterface

这个东西是在主函数中声明的变量，然后赋值给了app，网络接口最重要的一个变量是channelmap
这个变量管理着所有的socket连接。键就是地址。

Channel 就是socket连接

这个其实上面说过了，不过Channel的存放地方就是NetworkInterface

channel最重要的三个变量就是Sender，receiver和reader，reader会调用app或者entity里面定义的具体业务协议。sender是一个发送在发送不了的时候就会交给poller

bool TCPPacketSender::processSend(Channel* pChannel, int userarg)
{
	bool noticed = pChannel == NULL;

	// 如果是由poller通知的，我们需要通过地址找到channel
	if(noticed)
		pChannel = getChannel();

	KBE_ASSERT(pChannel != NULL);
	
	if(pChannel->condemn() == Channel::FLAG_CONDEMN_AND_DESTROY)
	{
		return false;
	}
	
	Channel::Bundles& bundles = pChannel->bundles();
	Reason reason = REASON_SUCCESS;

	Channel::Bundles::iterator iter = bundles.begin();
	for(; iter != bundles.end(); ++iter)
	{
		Bundle::Packets& pakcets = (*iter)->packets();
		Bundle::Packets::iterator iter1 = pakcets.begin();
		for (; iter1 != pakcets.end(); ++iter1)
		{
			reason = processPacket(pChannel, (*iter1), userarg);
			if(reason != REASON_SUCCESS)
				break; 
			else
				RECLAIM_PACKET((*iter)->isTCPPacket(), (*iter1));
		}

		if(reason == REASON_SUCCESS)
		{
			pakcets.clear();
			Network::Bundle::reclaimPoolObject((*iter));
			sendfailCount_ = 0;
		}
		else
		{
			pakcets.erase(pakcets.begin(), iter1);
			bundles.erase(bundles.begin(), iter);

			if (reason == REASON_RESOURCE_UNAVAILABLE)
			{
				/* 此处输出可能会造成debugHelper处死锁
					WARNING_MSG(fmt::format("TCPPacketSender::processSend: "
						"Transmit queue full, waiting for space(kbengine.xml->channelCommon->writeBufferSize->{})...\n",
						(pChannel->isInternal() ? "internal" : "external")));
				*/

				// 连续超过10次则通知出错
				if (++sendfailCount_ >= 10 && pChannel->isExternal())
				{
					onGetError(pChannel, "TCPPacketSender::processSend: sendfailCount >= 10");

					this->dispatcher().errorReporter().reportException(reason, pEndpoint_->addr(), 
						fmt::format("TCPPacketSender::processSend(external, sendfailCount({}) >= 10)", (int)sendfailCount_).c_str());
				}
				else
				{
					this->dispatcher().errorReporter().reportException(reason, pEndpoint_->addr(), 
						fmt::format("TCPPacketSender::processSend({}}, {})", (pChannel->isInternal() ? "internal" : "external"), (int)sendfailCount_).c_str());
				}
			}
			else
			{
				if (pChannel->isExternal())
				{
#if KBE_PLATFORM == PLATFORM_UNIX
					this->dispatcher().errorReporter().reportException(reason, pEndpoint_->addr(), "TCPPacketSender::processSend(external)",
						fmt::format(", errno: {}", errno).c_str());
#else
					this->dispatcher().errorReporter().reportException(reason, pEndpoint_->addr(), "TCPPacketSender::processSend(external)",
						fmt::format(", errno: {}", WSAGetLastError()).c_str());
#endif
				}
				else
				{
#if KBE_PLATFORM == PLATFORM_UNIX
					this->dispatcher().errorReporter().reportException(reason, pEndpoint_->addr(), "TCPPacketSender::processSend(internal)",
						fmt::format(", errno: {}, {}", errno, pChannel->c_str()).c_str());
#else
					this->dispatcher().errorReporter().reportException(reason, pEndpoint_->addr(), "TCPPacketSender::processSend(internal)",
						fmt::format(", errno: {}, {}", WSAGetLastError(), pChannel->c_str()).c_str());
#endif
				}

				onGetError(pChannel, fmt::format("TCPPacketSender::processSend: errno={}", kbe_lasterror()));
			}

			return false;
		}
	}

	bundles.clear();

	if(noticed)
		pChannel->onSendCompleted();

	return true;
}

NetworkInerface的结构

总结一下

App的网络模块是：App定义一大堆业务协议handler，App里面有一个NetworkInterface，主管网络连接的，初始化的时候就绑定了端口。App的EventDispatcher里面有一个EventPoller，主管收发数据操作。EventPoller里面一个输入操作集合inputnotificationhandler和一个输出操作集合outputnotificationHander，新的连接chanel来的时候需要在连接初始化的时候注册输入和输出操作。游戏心跳不停地调用poller的收发数据事件，当有连接有数据收的时候会调用连接注册的收的回调，发送当时就在连接里面发送，如果发送延迟则交给poller以后延迟发送。

怎么说呢，我感觉我也没有掰扯清楚，不过道理越说越清楚，就是绕一下大圈，最后调用到业务的协议处理。