qtcpmsgserver.cpp

BCB下的高效开发包。包含大并发的网络库

//---------------------------------------------------------------------------
#pragma hdrstop
#include "QTcpMsgServer.h"
#include "../QDebugLoger/QDebugLoger.h"
#include "../QNetBase/QNetBase.h"
#include "../QStdlib/QStdlib.h"
using namespace N_QTcpMsgServer;
//------------------------------------------------------------------------------
IQTcpMsgServer* __stdcall IQTcpMsgServer::Create(void)
{
        return new QTcpMsgServer;
}
//------------------------------------------------------------------------------
__stdcall QTcpMsgServer::QTcpMsgServer()
    : m_Listener(NULL)
    , m_Server(NULL)
{
        this->SetReadLastData(false);
}
//------------------------------------------------------------------------------
__stdcall QTcpMsgServer::~QTcpMsgServer()
{
        this->Stop();
}
//------------------------------------------------------------------------------
bool __stdcall QTcpMsgServer::Start(unsigned int Port, unsigned int MaxSession
    , const std::string& EndSign, unsigned int MaxPackage)
{
        if (this->IsPowerOn())
                return false;
        if (EndSign.empty())
                return false;
        //创建/启动监听
        if (Port > 0) {
                m_Listener = IQTcpListenDevice::Create();
                if (!m_Listener)
                        return false;
                if (!m_Listener->Start(Port)) {
                        m_Listener->Release();
                        m_Listener = NULL;
                        return false;
                }
        }else{
                //不使用Listen,使用客户模式
        }
        //创建服务器
        m_Server = IQAsynTcpServer::Create();
        if (!m_Server) {
                m_Listener->Release();
                m_Listener = NULL;
                return false;
        }
        //完成初始
        m_MaxSession    = MaxSession;
        m_EndSign       = EndSign;
        m_NewIter       = 1001;
        m_MaxPackage    = MaxPackage;
        m_Map.clear();
        m_Msg.clear();
        return true;
}
//------------------------------------------------------------------------------
void __stdcall QTcpMsgServer::Stop(void)
{
        if (!this->IsPowerOn())
                return;
        if (m_Listener) {
                m_Listener->Stop();
                m_Listener->Release();
                m_Listener = NULL;
        }

        m_Server->Release();
        m_Server = NULL;
        
        m_NewIter = 1001;
        m_Map.clear();
        m_Msg.clear();
}
//------------------------------------------------------------------------------
bool __stdcall QTcpMsgServer::IsPowerOn(void)const
{
        return NULL != m_Server;
}
//------------------------------------------------------------------------------
bool __stdcall QTcpMsgServer::Send(int SessionID, const std::string& Msg)
{
        if (!this->IsPowerOn())
                return false;
        ID_MAP::iterator iter = m_Map.find(SessionID);
        if (m_Map.end() == iter)
                return false;
                
        int Sock = m_Map[SessionID].Sock;
        assert (0 != Sock);
        return m_Server->Send(Sock, Msg);
}
//------------------------------------------------------------------------------
bool __stdcall QTcpMsgServer::GetQueuedMsg(int* pSessionID, std::string* pMsg)
{
        if (!pMsg || !pSessionID)
                return false;
        MSGITEM MsgItem;
        if (!m_Msg.recv(&MsgItem))
                return false;
        *pSessionID = MsgItem.SessionID;
        *pMsg       = MsgItem.Msg;
        return true;
}
//------------------------------------------------------------------------------
std::string __stdcall QTcpMsgServer::GetAddress(int SessionID)const
{
        if (!this->IsPowerOn())
                return "";
        const ID_MAP::const_iterator iter = m_Map.find(SessionID);
        if (m_Map.end() == iter)
                return "";
                
        int Sock = iter->second.Sock;
        assert (0 != Sock);
        QConn conn = _QGetRemote(Sock);
        return conn.IpAddr + ":" + QIntToStr(conn.Port);
}
//------------------------------------------------------------------------------
void __stdcall QTcpMsgServer::ShutDown(int SessionID)
{
        if (!this->IsPowerOn())
                return;
        ID_MAP::iterator iter = m_Map.find(SessionID);
        if (m_Map.end() == iter)
                return;
                
        int Sock = iter->second.Sock;
        if (0 == Sock)
                return;
        if (m_Server->GetSendStore(Sock) > 1024 * 1024) {
                m_Server->ShutDownForce(Sock);//需要发送的信息过多，应当立即销毁
                iter->second.Buf.clear();//清空这里,以便能被recv检测到断开,然后被Process销毁
                return;
        }
        m_Server->ShutDown(Sock);//此时是发送信息完成后销毁
}
//------------------------------------------------------------------------------
void __stdcall QTcpMsgServer::Process(void)
{
        if (!this->IsPowerOn())
                return;
        if (m_Listener) {
                m_Listener->Process();
        }
        this->ProcessAccept();


        this->ProcessDelete();
        this->ProcessRecv();
        //QDEBUG_LOG("m_Server->Process();", 1);
        m_Server->Process();
}
//------------------------------------------------------------------------------
void __stdcall QTcpMsgServer::SetReadLastData(bool IsReadLast)
{
        m_IsReadLastData = IsReadLast;
}
//------------------------------------------------------------------------------
void __stdcall QTcpMsgServer::ProcessDelete(void)
{
        //QDEBUG_LOG("PROCESS DELETE", m_Server->GetCount());
        ID_MAP::iterator iter = m_Map.begin();
        const ID_MAP::iterator iend = m_Map.end();
        std::vector<int> ReadyToDelete;
        for(; iter != iend; ++iter) {
                const unsigned int SessionID = iter->first;
                ITEM& Item = iter->second;//注意是引用
                if (!m_Server->IsExist(Item.Sock)) {
                        //QDEBUG_LOG("NOT IsExist", Item.Sock);
                        continue;
                }
                if (m_Server->IsAlive(Item.Sock))
                        continue;
                //QDEBUG_LOG("NOT ALIVE", Item.Sock);
                ReadyToDelete.push_back(SessionID);//索引
        }

        for(unsigned int i = 0; i < ReadyToDelete.size(); ++i) {
                this->Remove(ReadyToDelete[i]);
        }
}
//------------------------------------------------------------------------------
void __stdcall QTcpMsgServer::ProcessRecv(void)
{
        //QDEBUG_LOG("PROCESS RECV", m_Server->GetCount());
        ID_MAP::iterator iter = m_Map.begin();
        const ID_MAP::iterator iend = m_Map.end();
        std::string Data;
        for(; iter != iend; ++iter) {
                const unsigned int SessionID = iter->first;
                ITEM& Item = iter->second;//注意是引用

                if (!m_Server->IsExist(Item.Sock)) {
                        //QDEBUG_LOG("NOT IsExist", Item.Sock);
                        continue;
                }
                if (!m_Server->IsAlive(Item.Sock)) {
                        //QDEBUG_LOG("不可能的流程2", Item.Sock);
                        continue;
                }
                //QDEBUG_LOG("ALIVE", Item.Sock);

                Data.clear();
                //QDEBUG_LOG("CLEAR To RECVing", Item.Sock);
                m_Server->Recv(Item.Sock, &Data);
                //QDEBUG_LOG("RECV END OUT", Item.Sock);
                if (Data.empty()) {
                        //QDEBUG_LOG("NO DATA", Item.Sock);
                        continue;
                }
                //QDEBUG_LOG("DATA", Item.Sock);
                Item.Buf += Data;
                if (std::string::npos == Item.Buf.find(m_EndSign)) {
                        //检测非完整消息是否超长
                        if (Item.Buf.size() > m_MaxPackage) {
                                m_Server->ShutDown(Item.Sock);
                        }
                        continue;
                }
                //QDEBUG_LOG("MSG", Item.Sock);
                while(std::string::npos != Item.Buf.find(m_EndSign)) {
                        MSGITEM MsgItem;
                        MsgItem.Msg = GetNext(Item.Buf, m_EndSign);
                        MsgItem.SessionID = SessionID;//保持 MsgItem.Buf 为空
                        m_Msg.send(MsgItem);

                }
        }
}
//------------------------------------------------------------------------------
void __stdcall QTcpMsgServer::ProcessGC(void)
{
        m_NewIter = 1001;
}
//------------------------------------------------------------------------------
void __stdcall QTcpMsgServer::ProcessAccept(void)
{
        if (!m_Listener)
                return;
        int NewSock = 0;
        while (m_Listener->GetAccepted(&NewSock)) {
                if (!this->Append(NewSock)) {
                        _QClose(NewSock);
                        continue;
                }
        }
        
}
//------------------------------------------------------------------------------
bool __stdcall QTcpMsgServer::Append(int Sock) //警告:不可更改m_NewIter运动方式,PushSocket受依赖
{
        if (m_Server->GetCount() != m_Map.size()) {//同步检测
                MsgBox("不同步","bool QTcpMsgServer::Append(int Sock)");
        }
        if (m_Server->GetCount() >= m_MaxSession)
                return false;
                
        if (!m_Server->Insert(Sock))
                return false;

        while(m_Map.end() != m_Map.find(m_NewIter)) {
                if (m_NewIter > 0x7FFFFFF0)
                        return false;//应该使用ProcessGC来回滚
                ++m_NewIter;
        }
        //现在m_NewIter是一个新号

        ITEM Item;
        Item.Sock = Sock;
        m_Map[m_NewIter] = Item;
        ++ m_NewIter;//一定不要在最后一个号码上停留
        return true;
}
//------------------------------------------------------------------------------
int  __stdcall QTcpMsgServer::PushSocket(int Sock)
{
        if (!this->IsPowerOn())
                return 0;
        if (this->Append(Sock)) {
                return m_NewIter - 1;
        }
        return 0;
}
//------------------------------------------------------------------------------
bool __stdcall QTcpMsgServer::IsExist(int SessionID)const
{
        return m_Map.end() != m_Map.find(SessionID);
}
//------------------------------------------------------------------------------
void __stdcall QTcpMsgServer::Remove(int SessionID)
{
        //QDEBUG_LOG("TO Remove", 0);
        if (!this->IsExist(SessionID))//保证了不会重复删除并出现重复队列消息
                return;
        //如果允许则发送最后剩余数据
        if (m_IsReadLastData) {
                MSGITEM LastDataItem;
                LastDataItem.SessionID = SessionID;
                LastDataItem.Msg       = m_Map[SessionID].Buf;
                if (!LastDataItem.Msg.empty())
                        m_Msg.send(LastDataItem);
        }

        //开始删除
        m_Server->Delete(m_Map[SessionID].Sock);
        m_Map.erase(SessionID);
        MSGITEM MsgItem;
        MsgItem.SessionID = SessionID;//保持 MsgItem.Buf 为空
        m_Msg.send(MsgItem);
        //QDEBUG_LOG("Removed", 0);
}
//------------------------------------------------------------------------------
void __stdcall QTcpMsgServer::SetTimeOut(unsigned int TimeOut)
{
        m_Server->SetTimeOut(TimeOut);
}
//------------------------------------------------------------------------------
unsigned int __stdcall QTcpMsgServer::GetCount(void)const
{
        return m_Map.size();
}
//------------------------------------------------------------------------------