Ok...forgot a couple of things. Bu::ItoServer now monitors all client

connections and actually cleans up behind them when they're all done.  Seems
important.  It also will cleanup any lingering sockets that are laying around
at destruction time, although right now unless you force-stop the server thread
there is no way to interrupt it.  That'll come in a bit.

2 files changed, 41 insertions, 52 deletions

diff --git a/src/itoserver.cpp b/src/itoserver.cpp
index 0337057..1bad872 100644
--- a/src/itoserver.cpp
+++ b/src/itoserver.cpp
@@ -14,6 +14,18 @@ Bu::ItoServer::ItoServer() :
 
 Bu::ItoServer::~ItoServer()
 {
+        while( !qClientCleanup.isEmpty() )
+        {
+                ItoClient *pCli = qClientCleanup.dequeue();
+                pCli->join();
+                delete pCli;
+        }
+        for( ClientHash::iterator i = hClients.begin(); i != hClients.end(); i++ )
+        {
+                ItoClient *pCli = (*i);
+                pCli->join();
+                delete pCli;
+        }
 }
 
 void Bu::ItoServer::addPort( int nPort, int nPoolSize )
@@ -37,58 +49,16 @@ void Bu::ItoServer::setTimeout( int nTimeoutSec, int nTimeoutUSec )
         this->nTimeoutSec = nTimeoutSec;
         this->nTimeoutUSec = nTimeoutUSec;
 }
-/*
-void Bu::ItoServer::scan()
-{
-        struct timeval xTimeout = { nTimeoutSec, nTimeoutUSec };
-        
-        fd_set fdRead = fdActive;
-        fd_set fdWrite = fdActive;
-        fd_set fdException = fdActive;
-
-        if( TEMP_FAILURE_RETRY( select( FD_SETSIZE, &fdRead, NULL, &fdException, &xTimeout ) ) < 0 )
-        {
-                throw ExceptionBase("Error attempting to scan open connections.");
-        }
 
-        for( int j = 0; j < FD_SETSIZE; j++ )
-        {
-                if( FD_ISSET( j, &fdRead ) )
-                {
-                        if( hServers.has( j ) )
-                        {
-                                ServerSocket *pSrv = hServers.get( j );
-                                addClient( pSrv->accept(), pSrv->getPort() );
-                        }
-                        else
-                        {
-                                Client *pClient = hClients.get( j );
-                                pClient->processInput();
-                                if( !pClient->isOpen() )
-                                {
-                                        onClosedConnection( pClient );
-                                        hClients.erase( j );
-                                        FD_CLR( j, &fdActive );
-                                }
-                        }
-                }
-        }
-
-        // Now we just try to write all the pending data on all the sockets.
-        // this could be done better eventually, if we care about the socket
-        // wanting to accept writes (using a select).
-        for( ClientHash::iterator i = hClients.begin(); i != hClients.end(); i++ )
-        {
-                (*i)->processOutput();
-        }
-}
-*/
 void Bu::ItoServer::addClient( int nSocket, int nPort )
 {
         ItoClient *pC = new ItoClient( *this, nSocket, nPort, nTimeoutSec,
                         nTimeoutUSec );
         pC->start();
-        
+
+        imClients.lock();
+        hClients.insert( nSocket, pC );
+        imClients.unlock();
 }
 
 void *Bu::ItoServer::run()
@@ -114,11 +84,26 @@ void *Bu::ItoServer::run()
                                 addClient( pSrv->accept(), pSrv->getPort() );
                         }
                 }
+
+                while( !qClientCleanup.isEmpty() )
+                {
+                        ItoClient *pCli = qClientCleanup.dequeue();
+                        pCli->join();
+                        delete pCli;
+                }
         }
 
         return NULL;
 }
 
+void Bu::ItoServer::clientCleanup( int iSocket )
+{
+        imClients.lock();
+        ItoClient *pCli = hClients.get( iSocket );
+        imClients.unlock();
+        qClientCleanup.enqueue( pCli );
+}
+
 Bu::ItoServer::ItoClient::ItoClient( ItoServer &rSrv, int iSocket, int iPort,
                 int nTimeoutSec, int nTimeoutUSec ) :
         rSrv( rSrv ),
@@ -133,7 +118,6 @@ Bu::ItoServer::ItoClient::ItoClient( ItoServer &rSrv, int iSocket, int iPort,
         pClient = new Client(
                 new Bu::Socket( iSocket )
                 );
-
 }
 
 Bu::ItoServer::ItoClient::~ItoClient()
@@ -163,6 +147,7 @@ void *Bu::ItoServer::ItoClient::run()
                         {
                                 rSrv.onClosedConnection( pClient );
 
+                                rSrv.clientCleanup( iSocket );
                                 return NULL;
                         }
                 }
diff --git a/src/itoserver.h b/src/itoserver.h
index 19c34ca..60b15b4 100644
--- a/src/itoserver.h
+++ b/src/itoserver.h
@@ -7,6 +7,7 @@
 #include "bu/list.h"
 #include "bu/ito.h"
 #include "bu/itomutex.h"
+#include "bu/itoqueue.h"
 #include "bu/set.h"
 
 namespace Bu
@@ -36,6 +37,7 @@ namespace Bu
          */
         class ItoServer : public Ito
         {
+                friend class ItoClient;
         public:
                 ItoServer();
                 virtual ~ItoServer();
@@ -78,11 +80,13 @@ namespace Bu
                 fd_set fdActive;
                 typedef Hash<int,ServerSocket *> ServerHash;
                 ServerHash hServers;
-                //typedef Bu::Set<ItoClient *> ClientSet;
-                //ClientSet sClients;
-                //typedef Hash<int,Client *> ClientHash;
-                //ClientHash hClients;
-                ItoMutex im;
+                typedef Hash<int,ItoClient *> ClientHash;
+                typedef ItoQueue<ItoClient *> ClientQueue;
+                ClientHash hClients;
+                ClientQueue qClientCleanup;
+                ItoMutex imClients;
+
+                void clientCleanup( int iSocket );
         };
 }