tcpserver.cpp

C++ patterns设计模式

#include "stdafx.h"

#include "tcpserver.h"

#ifdef _DEBUG
#define new DEBUG_NEW
#endif

namespace acl
{
    using std::string;

    static const int WAIT_EXIT_TIME = 5000;
    static const int REPARE_TIME	= 3000;
      
    TcpServer::TcpServer()
    {
        init();
    }
    
    TcpServer::TcpServer(u_short port, const char *addr/* = NULL*/)
        : m_comm(SocketAddress(addr, port))
   {
        init();
    }
    
    TcpServer::~TcpServer()
    {
        stop(WAIT_EXIT_TIME);
    }

    void TcpServer::init()
    {
        m_running   = false;
        m_binded    = false;
        m_exit_signaled = false;
        m_clients   = 0;
    }
    
    bool TcpServer::start()
    {
        if(!m_running)
        {            
            if(m_comm.bind())
            {
                m_running = true;
                m_binded = true;
                startListen();
            }
            else
            {
                trace(0, "ERR:bind failed, error(%d)\n", m_comm.getError());
            }
        }
        return m_running;
    }
    
    bool TcpServer::start(u_short port, const char *addr/* = NULL*/)
    {
        if(!m_running)
        {
            if(m_comm.bind(SocketAddress(addr, port)))
            {
                m_running = true;
                m_binded = true;
                startListen();
            }
            else
            {
                trace(0, "ERR:bind failed, error(%d)\n", m_comm.getError());
            }
        }
        return m_running;
    }
    
    void TcpServer::startListen()
    {
        m_exit_signaled = false;
        startThread();
    }
    
    void TcpServer::stop(int timeout/* = INFINITE*/)
    {
        if(m_running)
        {
            m_exit_signaled = true;
            //force the select action failed and return;
            m_comm.close();
            //wait for the thread finish completely and then we can exit fairily!
            waitFinish(timeout);
            stopThread();
            
            m_running = false;
            trace(0, "clean client connection...\n");
            cleanClients();
            trace(0, "clean finish\n");
            trace(0, "listen thread stopped\n");
            
            // reset runing stat 
            init();
        }
    }
    
	bool TcpServer::closeClient(int clientId)
	{
		AcceptedSocket client = getClientSocket(clientId);
		client.close();
		return true;
	}

    int TcpServer::send(unsigned int clientid, const char *buf, int buf_size)
    {
        AcceptedSocket client = getClientSocket(clientid);
        if(client.isValid())
        {
            traceBinary(clientid, "action send following buffer:\n", buf, buf_size);
            return client.sendAll(buf, buf_size);
        }
        else
        {
            trace(0, "ERR:can't find the spcecial client id(0x%X)\n", clientid);
            return SOCKET_ERROR;
        }
    }
    
    void TcpServer::makeSelector()
    {
        FD_ZERO(&m_readfds);
        //this server
        m_comm.setSelect(&m_readfds);

        for(int i=0; i<m_clients; ++i)
        {
            m_client[i].m_socketid.setSelect(&m_readfds);
        }        
        trace(0, "(%d)client(s) will be listen\n", m_clients);
    }
    
    void TcpServer::recover()
    {
        trace(0, "server recover begin ...\n");
        //close all connection
        cleanClients();
        trace(0, "server recover end\n");
    }
    
    //check the errcode and return true if that error reparable.
    bool TcpServer::reparable(int errcode)
    {
        return false;
    }
    
    bool TcpServer::insertClient(const AcceptedSocket &client)
    {
        if(m_clients < MAX_CLIENT)
        {
            m_client[m_clients].m_clientid = m_seqGenerator.next();
            m_client[m_clients].m_socketid = client;

			if(!onAccept(m_client[m_clients]))
			{
				trace(m_client[m_clients].getId(), "user not accept\n");
				return false;
			}
            trace(m_client[m_clients].getId(), "No.(%d) created, socketid(0x%X)\n",
                m_clients,
                m_client[m_clients].getsocket());
            ++m_clients;
            return true;
        }
        return false;
    }
    
    void TcpServer::eraseClient(int index)
    {
        if(index <0 || index > m_clients)
            return ;
        
        trace(m_client[index].getId(), "No.(%d)deleted, socketid(0x%X)\n",
										  index, 
                                          m_client[index].getsocket());
        
        onBroken(m_client[index]);
        m_client[index].m_socketid.close();
        //save the last value into i's position
        //the last position will be fill by other insert action
        //using the operator=
        if(index != (--m_clients))
            m_client[index] = m_client[m_clients];
    }
    
    void TcpServer::cleanClients()
    {
        for(int i=0; i< m_clients; ++i)
        {
            trace(m_client[i].getId(), "No.(%d)deleted, socketid(0x%X)\n",
                i, 
                m_client[i].getsocket());
            onBroken(m_client[i]);
            m_client[i].m_socketid.close();
        }
        //reset client number counter
        m_clients = 0;
    }
    
    AcceptedSocket TcpServer::getClientSocket(unsigned int clientid)
    {
        for(int i=0; i<m_clients; ++i)
        {
            if(m_client[i].m_clientid == clientid)
                return m_client[i].m_socketid;
        }
        // not find socket, just return an invalid socket
        return AcceptedSocket();
    }
    
    //the main job of this thread
    void TcpServer::threadProcess()
    {
        int ret;
        trace(0, "handler running now\n");
        for(;;)
        {
#ifndef _DEBUG
			try
			{
#endif // _DEBUG
				// get the select socket list
				makeSelector();
				ret = ::select(0, &m_readfds, NULL, NULL, NULL);
				// we check the exit flag here, coz we block berore
				if(m_exit_signaled)
				{
					break;
				}
				if(ret > 0)
				{
					//check accept request
					if(m_comm.isSelected(&m_readfds))
					{
						AcceptedSocket client = m_comm.accept();
						if(client.isValid())
						{
							if(!insertClient(client))
							{
								trace(0, "insert client failed\n");
								client.close();
							}
						}
						else
						{
							trace(0, "accept client failed, error(%d)\n", m_comm.getError());
						}
						continue;
					}
					//check all clients
					else
					{
						checkClientEvent();
					}
				}
				else
				{
					trace(0, "select failed, error(%d)\n", m_comm.getError());
					//recover the server's error status
					recover();
					continue ;
				}
#ifndef _DEBUG
			}
			catch(...)
			{
				trace(0, "tcpserver thread catch exception\n");
			}
#endif // _DEBUG
		}
        trace(0, "handler exit now\n");
    }
    
    void TcpServer::checkClientEvent()
    {
        char *buffer = NULL;
        int  length	 = -1;
        int  ret;
        int  closed_count = 0;
        int  closed_client[MAX_CLIENT] = {0};
        for(int i=0; i< m_clients; ++i)
        {
            if(m_client[i].m_socketid.isSelected(&m_readfds))
            {
				//RECV_OK             // recv ok
				//RECV_CLOSE          // connection closed
				//RECV_SOCKET_ERROR   // socket error
				//RECV_USER_ERROR     // user error < 0
                ret = onReceive(m_client[i]);
                switch(ret)
                {
                case RECV_OK:
                    {
                        trace(m_client[i].getId(), "buffer received\n");
                    }
                    break;
                case RECV_CLOSE:
                    {
                        trace(m_client[i].getId(), "connection close by peer\n");
                        closed_client[closed_count++] = i;
                    }
                    break;
                case RECV_SOCKET_ERROR:
                    {
                        trace(m_client[i].getId(), "WRN:handler occur network error(%d)\n", m_comm.getError());
                        closed_client[closed_count++] = i;
                    }
                    break;
                default:	//user failed
                    {
                        //check whether the error reparable
                        if(reparable(ret))
                        {
                            trace(m_client[i].getId(), "WRN:handler occur reparable error(0x%X)\n", ret);
                        }
                        else
                        {
                            //if unreparable then close the peer connection
                            trace(m_client[i].getId(), "WRN:handler occur unreparable error(0x%X)\n", ret);
                            closed_client[closed_count++] = i;
                        }
                    }
                    break;
                }
            }
        }
        //remove all closed client from the client array
        for(int j=0; j<closed_count; ++j)
        {
            eraseClient(closed_client[j]);
        }
    }
}