cdev-1.7.2n/extensions/cdevGenericServer/cdevClient/ServerHandler.cc

#include <cdevClock.h>
#include <ServerInterface.h>
#include <xdrClass.h>
#include <SignalManager.h>
#include <clipMagicNumber.h>

#define _CDEV_CONNECTION_RETRIES 5

SignalManager ServerHandler::signalManager;
int           ServerHandler::signalManagerInit = 0;

// *****************************************************************************
// * ServerHandlerCallbackNode:
// *	This is a node of a linked list that isolates the ServerHandlerCallback
// *	object from inadverntently being corrupted by being placed in more
// *	than one ServerHandler object.
// *
// *	The ALLOCATION_COUNT of 128 forces the allocation of blocks that are
// *	close to 1 kilobyte in size.
// *****************************************************************************
class ServerHandlerCallbackNode 
{
friend class ServerHandler;
private:
	static ServerHandlerCallbackNode * freeList_;
	ServerHandlerCallbackNode        * next;
	ServerHandlerCallback            * callback;
	enum { ALLOCATION_COUNT=128 }; 

	ServerHandlerCallbackNode ( void );
public:
	ServerHandlerCallbackNode ( ServerHandlerCallback * Callback );
	~ServerHandlerCallbackNode( void );
	void * operator new       ( size_t );
	void   operator delete    ( void * ptr );
};

ServerHandlerCallbackNode * ServerHandlerCallbackNode::freeList_;

// *****************************************************************************
// * ServerHandler::ServerHandler
// *	Default constructor for the ServerHandler class
// *****************************************************************************
ServerHandler::ServerHandler ( char * Server, ServerInterface * Interface ) 
	: SocketReader(CLIP_MAGIC_NUMBER), SocketWriter(CLIP_MAGIC_NUMBER),
	  queue(*(Interface->getQueue(Server))), serverface(Interface), callbacks(NULL), 
	  enqueuedPackets(0), enqueuedBytes(0), clientID(-1), contextID(-1),
	  tagChanged(1), serverQuitFlag(0)
	{
	strncpy(server, Server, 256);
	server[255]   = 0;	
	hostName[0]  = 0;
	clientID = getNextClientID();

	if(!signalManagerInit) signalManager.installDefaults();
	
	// *********************************************************************
	// * If any data exists in the queue for this server, walk through each
	// * entry and determine if it is valid - if so, reenqueue it, 
	// * otherwise, delete it.
	// *********************************************************************
	if(!queue.empty())
		{
		FifoQueue tempQueue;
		char * binary;
		size_t len;
		while(queue.dequeue(&binary, &len)==0)
			{
			if(serverface->isPacketValid(binary, len))
				{
				tempQueue.enqueue(binary, len);
				}
			else delete binary;
			}
		while(tempQueue.dequeue(&binary, &len)==0)
			{
			enqueue(binary, len);
			}
		}
	}

// *****************************************************************************
// * ServerHandler::outputError :
// *	This mechanism is used to report errors to the system.
// *****************************************************************************
int ServerHandler::outputError(int severity, char *name, char *formatString, ...)
	{
	va_list     argp;
	static char msg[1024];

	va_start (argp, formatString);
	vsprintf (msg, formatString, argp);
	va_end   (argp);

	return serverface->outputError(severity, name, msg);
	}
	
// *****************************************************************************
// * open:
// *	Established a connection with the host specified by remote_sap
// *****************************************************************************
int ServerHandler::open ( const cdevAddr & addr )
	{
	int result = 0;
	int count = 0;
	
	result = stream.connect (addr);

	if (result != -1)
		{
		if((result = serverface->Reactor.registerHandler(this, READ_MASK|WRITE_MASK|EXCEPT_MASK))==-1)
			{
			handleClose();
			}		
		}
	return result;
	}


// *****************************************************************************
// * ~ServerHandler:
// *	Default destructor for the ServerHandler object.
// *****************************************************************************
ServerHandler::~ServerHandler (void)
	{
	ServerHandlerCallbackNode * cb = callbacks;

	// *********************************************************************
	// * Write a -1 as the packetLength to the connection in order to 
	// * terminate.
	// *********************************************************************
	if(stream.getHandle()>0 && serverQuitFlag==0)
		{
		#ifndef _WIN32
			void (*action)(int) = signal(SIGPIPE, SIG_IGN);
		#endif
		
		writeGoodbye();
		
		#ifndef _WIN32
			signal(SIGPIPE, action);
		#endif
		}
		
	// *********************************************************************
	// * Contact all of the objects that rely on the existance of this
	// * class.
	// *********************************************************************
	while(cb!=NULL)
		{
		ServerHandlerCallbackNode * oldCB = cb;
		cb = cb->next;
		oldCB->callback->executeServerHandlerCallback(this);
		delete oldCB;
		}
		
	// *********************************************************************
	// * Perform standard shutdown steps.
	// *********************************************************************
	if(reactor) reactor->extractHandler(this);
	handleClose();
	}


// *****************************************************************************
// * ServerHandler::getHostName:
// *	This function returns the name of the remote host.
// *****************************************************************************
char * ServerHandler::getHostName( void )
	{
	if(*hostName==0)
		{
		cdevInetAddr addr;		
		if(stream.getRemoteAddress (addr)==0)
			{
			const char * ptr=addr.getHostName();
			if(ptr) strncpy (hostName, ptr, MAXHOSTNAMELEN+1);
			}
		}
	return hostName;
	}
	
// *****************************************************************************
// * ServerHandler::getName :
// * 	Allows the caller to obtain the name of the class.
// *****************************************************************************
char * ServerHandler::getName (void) const
	{
	return "ServerHandler"; 
	}


// *****************************************************************************
// * ServerHandler::getHandle
// * 	Allows the caller to obtain the file identifier of the socket.
// *****************************************************************************
int ServerHandler::getHandle (void) const 
	{
	return stream.getHandle(); 
	}



// *****************************************************************************
// * ServerHandler::getServer:
// * 	Allows the caller to obtain the name of the connected server.
// *****************************************************************************
char * ServerHandler::getServer (void) const
	{
	return (char *)server; 
	}


// *****************************************************************************
// * handleClose:
// * 	Closes the stream.
// *****************************************************************************
int ServerHandler::handleClose (void)
	{
	serverface->disconnect(server);
	stream.close();
	return 0;
	}


// *****************************************************************************
// * handleInput :
// *	This function is called when data is ready to be read from a connected
// *	socket.  This function will read the data from the socket, and will then
// *	submit the buffer to the ServerInterface class for processing.
// *****************************************************************************
int ServerHandler::handleInput (void)
{
int        retval = 0;
char     * buf    = NULL;
size_t     len    = 0;
int        result = 1;

// *****************************************************************************
// * Record oriented semantics dictate that the length of the transmission
// * always preceeds the actual data.  Therefore, read the length of the
// * transmission into the len variable.
// *****************************************************************************
while(result>0)	
	{
	result = read(&buf, (int *)&len);

	switch(result)
		{
		// *************************************************************
		// * A return value of SHUTDOWN_CODE indicates that a negative 
		// * one was provided as the packet length - indicating a 
		// * shutdown...
		// *************************************************************
		case SHUTDOWN_CODE:
		outputError(CDEV_SEVERITY_INFO, (char *)getName(),
			    "Connection to %s terminated by client", getHostName());      
		retval = -1;
		break;
	
		// *************************************************************
		// * A return value of -1 indicates an error occured while 
		// * reading from the socket.  A value of -1 is returned to 
		// * shutdown the socket and remove it from the reactor.
		// *************************************************************
		case -1:
		outputError(CDEV_SEVERITY_WARN, (char *)getName(),
			    "Error reading from connection to client %s", getHostName());      
		retval = -1;
		break;
	
		// *************************************************************
		// * A return value of 0 means that no data was ready to be 
		// * retrieved from the socket.
		// *************************************************************
		case 0:
		break;
		
		// *************************************************************
		// * Any other value returned from the socket represents the 
		// * number of bytes actually read into the local Session object.
		// *************************************************************
		default:
		serverface->enqueue(ServerInterface::COMPLETED, this, buf, len);
		break;		
       		}
	}

// *****************************************************************************
// * Return the value specified by retval.
// *****************************************************************************
return retval;
}



// *****************************************************************************
// * handleOutput :
// *	This function is called when data is ready to be transmitted to a 
// *	connected socket.  This function will read the data from the queue, 
// *	translate it to XDR, and then transmit it to the client.
// *****************************************************************************
int ServerHandler::handleOutput    ( void )
{
int                retval = 0;
char             * buf    = NULL;
size_t             len    = 0;

// *****************************************************************************
// * Attempt to transmit or continue transmitting the data.  Note, the peek
// * method is used to attempt to writeEnqueue a data item without actually
// * removing it from the outbound queue.  If the item can be writeEnqueued, 
// * then the dequeue method is called to remove it from the queue...
// * This method is repeated until the output buffer is fully populated.
// *****************************************************************************
if(!writing() && dequeue(&buf, &len)==0)
	{
	int full = 0;
	while(!full)
		{
		if(!(full = writeEnqueue(buf, len)))
			{
			delete buf;
			full = dequeue(&buf, &len);
			}
		else undequeue(buf, len);
		}
	if(writeContinue()<0) retval = -1;
	}
else if(writing())
	{
	if(writeContinue()<0) retval = -1;
	}

// *****************************************************************************
// * Display an error message if the transmission failed.
// *****************************************************************************
if(retval!=0) 
	{
	outputError(CDEV_SEVERITY_WARN, (char *)getName(),
		    "Error transmitting to %s", getHostName());
	}

// *****************************************************************************
// * If there are no further messages in the outbound queue and the
// * ACE_Event_Handler has finished servicing the current message, turn off the
// * WRITE_MASK for this ACE_Event_Handler.
// *****************************************************************************
if(empty() && !writing()) 
	{
	setMask(READ_MASK|EXCEPT_MASK);
	}
	
// *****************************************************************************
// * Return the value specified by retval.
// *****************************************************************************
return retval;
}


// *****************************************************************************
// * handleExcept :
// *	This function is called when out of band data has been received from 
// *	the server.  It will terminate the connection with the server.
// *****************************************************************************
int ServerHandler::handleExcept    ( void )
	{
	stream.close();
	serverQuitFlag=1;
	return -1;
	}


// *****************************************************************************
// * registerServerCallback :
// *	This method is called to add a ServerHandlerCallback object to the list
// *	of objects that the ServerHandler will notify when it is destroyed...
// *	This notification process protects objects that use this class for
// *	communications from using a pointer to a ServerHandler that has been
// *	deleted.
// *****************************************************************************
void ServerHandler::registerServerCallback(ServerHandlerCallback * cb)
	{
	ServerHandlerCallbackNode *cbPrev=NULL, *cbNext=callbacks;

	while(cbNext!=NULL && cbNext->callback!=cb)
		{
		cbPrev = cbNext;
		cbNext = cbNext->next;
		}
	if(cbNext==NULL)
		{
		if(cbPrev==NULL) callbacks = new ServerHandlerCallbackNode(cb);
		else cbPrev->next = new ServerHandlerCallbackNode(cb);
		}
	}

// *****************************************************************************
// * unregisterServerCallback :
// *	This method is used to remove a ServerHandlerCallback object from the
// *	list of objects that will be notified when the ServerHandler object
// *	is destroyed.  
// *****************************************************************************
void ServerHandler::unregisterServerCallback (ServerHandlerCallback * cb)
	{
	ServerHandlerCallbackNode *cbPrev=NULL, *cbNext=callbacks;

	while(cbNext!=NULL && cbNext->callback!=cb)
		{
		cbPrev = cbNext;
		cbNext = cbNext->next;
		}
	if(cbNext->callback==cb)
		{
		if(cbPrev==NULL) callbacks = cbNext->next;
		else cbPrev->next = cbNext->next;
		delete cbNext;
		}
	}
	

// *****************************************************************************
// * enqueue :
// *	This method places an object into the embedded FifoQueue and will force
// *	a flush of the socket if it reaches the high water mark.  
// *****************************************************************************
void ServerHandler::enqueue   (char * buf, size_t  len)
	{
	enqueuedPackets++;
	enqueuedBytes+=len;
	queue.enqueue(buf, len); 
	if(enqueuedPackets>100 || enqueuedBytes>16000) 
		{
		serverface->flush(getHandle());
		enqueuedPackets=0;
		enqueuedBytes=0;
		}
	}


// *****************************************************************************
// * ServerHandler::getNextClientID :
// *	This method allows the caller to retrieve a unique clientID to be
// *	assigned to a client.  The nextClientID value is automatically 
// *	incremented.
// *****************************************************************************
short ServerHandler::getNextClientID ( void )
	{
	static short nextClientID = 0;
	
	nextClientID++;
	if(nextClientID<=0) nextClientID = 1;
	return nextClientID; 
	}

	
// *****************************************************************************
// * ServerHandlerCallbackNode::ServerHandlerCallbackNode :
// *	Default constructor for the ServerHandlerCallbackNode class.
// *****************************************************************************
ServerHandlerCallbackNode::ServerHandlerCallbackNode ( void )
	: callback(NULL), next(NULL)
	{
	}


// *****************************************************************************
// * ServerHandlerCallbackNode::ServerHandlerCallbackNode :
// *	Parameterized constructor for the ServerHandlerCallbackNode class.
// *****************************************************************************
ServerHandlerCallbackNode::ServerHandlerCallbackNode( ServerHandlerCallback * Callback ) 
	: callback(Callback), next(NULL)
	{
	}


// *****************************************************************************
// * ServerHandlerCallbackNode::~ServerHandlerCallbackNode :
// *	Destructor for the ServerHandlerCallbackNode class.
// *****************************************************************************
ServerHandlerCallbackNode::~ServerHandlerCallbackNode( void )
	{
	}


// *****************************************************************************
// * ServerHandlerCallbackNode::operator new :
// *	Gets a ServerHandlerCallbackNode instance from the freelist.
// *****************************************************************************
void * ServerHandlerCallbackNode::operator new ( size_t )
	{
	ServerHandlerCallbackNode * result = NULL;

	if(freeList_==NULL) 
		{
		freeList_ = ::new ServerHandlerCallbackNode[ALLOCATION_COUNT];
		for(int i=0; i<ALLOCATION_COUNT; i++) 
			{
			freeList_[i].next = 
				(i<(ALLOCATION_COUNT-1))?&freeList_[i+1]:(ServerHandlerCallbackNode *)NULL;
			}
		}
	if(freeList_!=NULL) 
		{
		result                = freeList_;
		freeList_             = result->next;
		result->next          = NULL;
		}
	return result;
	}


// *****************************************************************************
// * ServerHandlerCallbackNode::operator delete :
// *	Returns a ServerHandlerCallbackNode instance to the freelist.
// *****************************************************************************
void ServerHandlerCallbackNode::operator delete ( void * ptr )
	{
	ServerHandlerCallbackNode * node = (ServerHandlerCallbackNode *)ptr;
	if(node != NULL) 
		{
		node->next = freeList_;
		freeList_  = node;
		}
	}