// **********************************************************************
//
// Copyright (c) 1999
// Object Oriented Concepts, Inc.
// Billerica, MA, USA
//
// All Rights Reserved
//
// **********************************************************************

#include <OB/Basic.h>
#include <OB/Except.h>
#include <OB/Template.h>
#include <OB/TemplateI.h>
#include <OB/Declarations.h>
#include <OB/Any.h>
#include <OB/IOP.h>
#include <OB/Object.h>
#include <OB/IntRepMember.h>
#include <OB/ORB.h>
#include <OB/ImplRep.h>
#include <OB/BOA.h>
#include <OB/OCI_IIOP.h>
#include <OB/GIOP.h>
#include <OB/Reactor.h>
#include <OB/Util.h> // For the message viewer

#include <OCI_impl.h> // For OCI_Current_impl::_OB_...()

#include <GIOPServerStarter.h>
#include <GIOPServerWorker.h>

#ifdef HAVE_JTC
//
// Monitor for "single threaded" concurrency model
//
JTCMutex OBGIOPServerWorkerThreaded::singleThreadedMutex_;
#endif

#ifdef HAVE_JTC
OBGIOPServerWorkerThreaded::ThreadPool*
OBGIOPServerWorkerThreaded::ThreadPool::instance_ = 0;
JTCMutex OBGIOPServerWorkerThreaded::ThreadPool::instanceMutex_;
#endif

// ----------------------------------------------------------------------
// External, non-inline duplicate/release for templates
// ----------------------------------------------------------------------

void
OBDuplicate(OBGIOPServerWorker_ptr p)
{
    if(p)
	p -> _OB_incRef();
}

void
OBRelease(OBGIOPServerWorker_ptr p)
{
    if(p)
	p -> _OB_decRef();
}

void
OBDuplicate(OBGIOPServerWorkerBlocking_ptr p)
{
    if(p)
	p -> _OB_incRef();
}

void
OBRelease(OBGIOPServerWorkerBlocking_ptr p)
{
    if(p)
	p -> _OB_decRef();
}

void
OBDuplicate(OBGIOPServerWorkerReactive_ptr p)
{
    if(p)
	p -> _OB_incRef();
}

void
OBRelease(OBGIOPServerWorkerReactive_ptr p)
{
    if(p)
	p -> _OB_decRef();
}

#ifdef HAVE_JTC
void
OBDuplicate(OBGIOPServerWorkerThreaded_ptr p)
{
    if(p)
	p -> _OB_incRef();
}

void
OBRelease(OBGIOPServerWorkerThreaded_ptr p)
{
    if(p)
	p -> _OB_decRef();
}
#endif

// ----------------------------------------------------------------------
// Template instantiations
// ----------------------------------------------------------------------

#ifndef HAVE_NO_EXPLICIT_TEMPLATES
template class OBObjVar< OBGIOPServerWorker >;
template class OBObjForSeq< OBGIOPServerWorker >;
template class OBObjSeq< OBGIOPServerWorker >;
//template class OBObjVar< OBGIOPServerWorkerBlocking >;
//template class OBObjForSeq< OBGIOPServerWorkerBlocking >;
template class OBObjVar< OBGIOPServerWorkerReactive >;
template class OBObjForSeq< OBGIOPServerWorkerReactive >;
#ifdef HAVE_JTC
//template class OBObjVar< OBGIOPServerWorkerThreaded >;
//template class OBObjForSeq< OBGIOPServerWorkerThreaded >;
#endif
#else
#ifdef HAVE_PRAGMA_DEFINE
#pragma define(OBObjVar< OBGIOPServerWorker >)
#pragma define(OBObjForSeq< OBGIOPServerWorker >)
#pragma define(OBObjSeq< OBGIOPServerWorker >)
//#pragma define(OBObjVar< OBGIOPServerWorkerBlocking >)
//#pragma define(OBObjForSeq< OBGIOPServerWorkerBlocking >)
#pragma define(OBObjVar< OBGIOPServerWorkerReactive >)
#pragma define(OBObjForSeq< OBGIOPServerWorkerReactive >)
#ifdef HAVE_JTC
//#pragma define(OBObjVar< OBGIOPServerWorkerThreaded >)
//#pragma define(OBObjForSeq< OBGIOPServerWorkerThreaded >)
#endif
#endif
#endif

// ----------------------------------------------------------------------
// OBGIOPServerWorker private and protected member implementations
// ----------------------------------------------------------------------

//
// Display a warning when an unexpected exception is raised by a
// servant. In addition to an error message, the operation name and
// acceptor information is displayed, if possible.
//
void
OBGIOPServerWorker::displayWarning(const char* err,
				   const char* op,
				   OCI_TransportInfo_ptr transportInfo)
{
    CORBA_String_var msg = CORBA_string_dup(err);
    msg += "\noperation name=\"";
    msg += op;
    msg += "\" acceptor=";
    OCI_AcceptorInfo_var acceptorInfo = transportInfo -> acceptor_info();
    OCI_IIOP_AcceptorInfo_var iiopAcceptorInfo =
	OCI_IIOP_AcceptorInfo::_narrow(acceptorInfo);
    if(!CORBA_is_nil(iiopAcceptorInfo))
    {
	CORBA_String_var host = iiopAcceptorInfo -> host();
	CORBA_UShort port = iiopAcceptorInfo -> port();
	msg += host;
	msg += ":";
	msg += port;
    }
    else
    {
	msg += "not iiop";
    }
    OBMessageViewer::instance() -> warning(msg);
}

void
OBGIOPServerWorker::addMessageError()
{
    OCI_Buffer_var buf = new OCI_Buffer(12);
    CORBA_Octet* oct = buf -> data();
    GIOP_MessageHeader hdr =
    {
	{ 'G', 'I', 'O', 'P' }, { 1, 0 },
	OBEndian, GIOP_MessageError, 0
    };
    OBMarshal(hdr, oct);
    add(buf._retn());
}

void
OBGIOPServerWorker::addCloseConnection()
{
    OCI_Buffer_var buf = new OCI_Buffer(12);
    CORBA_Octet* oct = buf -> data();
    GIOP_MessageHeader hdr =
    {
	{ 'G', 'I', 'O', 'P' }, { 1, 0 },
	OBEndian, GIOP_CloseConnection, 0
    };
    OBMarshal(hdr, oct);
    add(buf._retn());
}

void
OBGIOPServerWorker::setupBuffer()
{
    //
    // Create new buffer for incoming message
    //
    buf_ = new OCI_Buffer(12);
}

void
OBGIOPServerWorker::extractHeader()
{
    const CORBA_Octet* coct = buf_ -> data();
    
    if(coct[0] != 'G' || coct[1] != 'I' || coct[2] != 'O' || coct[3] != 'P')
    {
	addMessageError();
	throw CORBA_COMM_FAILURE("", OBMinorNoGIOP, CORBA_COMPLETED_NO);
    }
    
    OBUnmarshal(hdr_, coct, false);
    assert(coct - buf_ -> data() == 12);
    bool swap = (hdr_.flag & 0x1) != OBEndian;
    CORBA_ULong sz;
    const CORBA_Octet* coctSz = (const CORBA_Octet*)&hdr_.message_size;
    OBUnmarshal(sz, coctSz, swap);
    hdr_.message_size = sz;
    buf_ -> realloc(12 + sz);
}

//
// The use of this method differs from the standard C++ CORBA mapping
// because add() takes ownership of the OCI_Buffer_ptr parameter. That
// is by passing a buffer to this method the caller relinquishes
// ownership. This is because we don't want to make the reference
// counting of the OCI_Buffer class thread-safe.
//
void
OBGIOPServerWorker::add(OCI_Buffer_ptr buf)
{
    //
    // Add new message to the message buffers
    //
    outSeq_.length(outSeq_.length() + 1);
    outSeq_[outSeq_.length() - 1] = buf;
}

void
OBGIOPServerWorker::sendRemainingMessages()
{
    //
    // Try to send all unsent outgoing messages
    //
    for(CORBA_ULong i = 0 ; i < outSeq_.length() ; i++)
    {
	while(!outSeq_[i] -> is_full())
	{
	    CORBA_ULong oldPos = outSeq_[i] -> pos();
	    
	    //
	    // Send buffer, non-blocking
	    //
	    try
	    {
          	OB_DEC_TRY_FIX
		transport_ -> send(outSeq_[i], false);
	    }
	    catch(const CORBA_SystemException&)
	    {
	    }

	    //
            // Don't allow any delay if messages cannot be send
	    //
	    if(outSeq_[i] -> pos() == oldPos)
	    {
		i = outSeq_.length();
		break;
	    }
	}
    }
}

void
OBGIOPServerWorker::closeTransport()
{
    //
    // `discard' parameter must be set to FALSE, as buffered messages,
    // such as CloseConnection messages, must be reliably transmitted.
    //
    transport_ -> close(CORBA_FALSE);
}

void
OBGIOPServerWorker::execute()
{
    //
    // Save buffer and all relevant information from the header
    //
    bool swap = (hdr_.flag & 0x1) != OBEndian;
    GIOP_MsgType type = (GIOP_MsgType)hdr_.message_type;
    OBBuffer buf;
    buf.consume(buf_ -> buf());
    buf_ = OCI_Buffer::_nil(); // Reset buf_ to allow for reading
                               // the next message
    
    //
    // Check message type
    //
    if(type > GIOP_MessageError)
    {
	addMessageError();
	throw CORBA_COMM_FAILURE("", OBMinorUnknownMessage,
				 CORBA_COMPLETED_NO);
    }
    
    switch(type)
    {
    case GIOP_Request:

	dispatchRequest(buf, swap);
	
	break;
    
    case GIOP_CancelRequest:
	
	//
	// Do nothing here (doing nothing is GIOP compliant)
	//
	break;
	
    case GIOP_LocateRequest:
    {
	//
	// Get the locate request header
	//
	GIOP_LocateRequestHeader locReqH;
	const CORBA_Octet* coct = buf.data + 12;
	OBUnmarshal(locReqH, coct, swap);
	
	//
	// Start creating locate reply message
	//
	GIOP_LocateReplyHeader locRepH;
	locRepH.request_id = locReqH.request_id;
	CORBA_ULong offsetOut = 12;
	OBMarshalCount(locRepH, offsetOut);
	
	//
	// Execute request, using the BOA
	//
	CORBA_BOA_ptr boa = CORBA_BOA::_OB_instance_nodup();
	CORBA_Object_var impl = boa -> _OB_find(locReqH.object_key);
	
	if(!CORBA_is_nil(impl))
	{
	    CORBA_ULong count = offsetOut;
	    /* No body to marshal here ... */
	    
	    buf.alloc(count);
	    locRepH.locate_status = GIOP_OBJECT_HERE;
	}
	else
	{
	    CORBA_ULong count = offsetOut;
	    /* No body to marshal here ... */
	    
	    buf.alloc(count);
	    locRepH.locate_status = GIOP_UNKNOWN_OBJECT;
	}
	
	//
	// Create locate reply message
	//
	GIOP_MessageHeader hdr =
	{
	    { 'G', 'I', 'O', 'P' }, { 1, 0 },
	    OBEndian, GIOP_LocateReply, buf.len - 12
	};
	
	CORBA_Octet* oct = buf.data;
	OBMarshal(hdr, oct);
	OBMarshal(locRepH, oct);
	
	//
	// Convert simple buffer to OCI buffer
	// Add the OCI buffer
	//
	OCI_Buffer_var b = new OCI_Buffer;
	b -> buf().consume(buf);
	add(b._retn());
	
	break;
    }
    
    case GIOP_Reply:
    case GIOP_LocateReply:
    case GIOP_CloseConnection:
	
	//
	// These messages may not be sent to a server
	//
	addMessageError();
	throw CORBA_COMM_FAILURE("", OBMinorWrongMessage,
				 CORBA_COMPLETED_NO);
	
    case GIOP_MessageError:
	
	//
	// How are message error messages handled? I just send an
	// error message back.
	//
	addMessageError();
	throw CORBA_COMM_FAILURE("", OBMinorMessageError,
				 CORBA_COMPLETED_NO);
    }
}
	
void
OBGIOPServerWorker::dispatchRequest(OBBuffer& buf, bool swap)
{
    //
    // Get the request header
    //
    GIOP_RequestHeader reqH;
    const CORBA_Octet* coct = buf.data + 12;
    OBUnmarshal(reqH, coct, swap);
    CORBA_ULong offsetIn = coct - buf.data;

    OCI_TransportInfo_var transportInfo = transport_ -> get_info();
    
    if(reqH.response_expected)
    {
	//
	// Start creating reply message
	//
	GIOP_ReplyHeader repH;
	repH.request_id = reqH.request_id;
	repH.reply_status = GIOP_NO_EXCEPTION;
	CORBA_ULong offsetOut = 12;
	OBMarshalCount(repH, offsetOut);

	//
	// Execute request, using the BOA
	//
	CORBA_BOA_ptr boa = CORBA_BOA::_OB_instance_nodup();
        
        try
        {
            //
            // We set the current OCI transport. We need to save the
            // transport so that we can restore the current transport
            // after the call.
            //
            OCI_Current_impl* current = OCI_Current_impl::_OB_impl();
            OCI_TransportInfo_var save =
                current -> _OB_setOCITransportInfo(transportInfo);

            OBDispatchStatus status;
            try
            {
		OB_DEC_TRY_FIX
                status = boa -> _OB_dispatch(reqH.object_key, reqH.operation,
                                             buf, swap, offsetIn, offsetOut);
            }
            catch(...)
            {
                //
                // We need to save the return value of to a _var type to
                // ensure that the reference is released.
                //
                OCI_TransportInfo_var dummy =
		    current -> _OB_setOCITransportInfo(save);
                throw;
            }

	    //
	    // We need to save the return value of to a _var type to
	    // ensure that the reference is released.
	    //
            OCI_TransportInfo_var dummy =
                current -> _OB_setOCITransportInfo(save);
	    
	    switch(status)
	    {
	    case OBDispatchStatusOK:
		break;
		
	    case OBDispatchStatusExcept:
		repH.reply_status = GIOP_USER_EXCEPTION;
		break;
		    
	    case OBDispatchStatusNoObject:
	    {
		CORBA_OBJECT_NOT_EXIST ex;
		CORBA_ULong count = offsetOut;
		OBMarshalCount(ex, count);
		
		buf.alloc(count);
		CORBA_Octet* oct = buf.data + offsetOut;
		OBMarshal(ex, oct);
		
		repH.reply_status = GIOP_SYSTEM_EXCEPTION;
		break;
	    }
		
	    case OBDispatchStatusNoOperation:
	    {
		CORBA_BAD_OPERATION ex;
		CORBA_ULong count = offsetOut;
		OBMarshalCount(ex, count);
		    
		buf.alloc(count);
		CORBA_Octet* oct = buf.data + offsetOut;
		OBMarshal(ex, oct);
		    
		repH.reply_status = GIOP_SYSTEM_EXCEPTION;
		break;
	    }
	    }
	}
	catch(const CORBA_SystemException& ex)
	{
	    CORBA_ULong count = offsetOut;
	    OBMarshalCount(ex, count);
		
	    buf.alloc(count);
	    CORBA_Octet* oct = buf.data + offsetOut;
	    OBMarshal(ex, oct);
		
	    repH.reply_status = GIOP_SYSTEM_EXCEPTION;
	}
        catch(const CORBA_UserException&)
	{
	    //
	    // Print a warning message
	    //
	    displayWarning(
		"Servant method raised an undeclared CORBA user exception\n"
		"Client receives this exception as CORBA::UNKNOWN",
		reqH.operation,
		transportInfo);

	    //
	    // Replace undeclared CORBA user exceptions with
	    // CORBA::UNKNOWN
	    //
	    CORBA_UNKNOWN ex;

	    CORBA_ULong count = offsetOut;
	    OBMarshalCount(ex, count);
		
	    buf.alloc(count);
	    CORBA_Octet* oct = buf.data + offsetOut;
	    OBMarshal(ex, oct);
		
	    repH.reply_status = GIOP_SYSTEM_EXCEPTION;
	}
        catch(...)
	{
	    //
	    // Print a warning message
	    //
	    displayWarning(
		"Servant method raised a non-CORBA exception\n"
		"Client receives this exception as CORBA::UNKNOWN",
		reqH.operation,
		transportInfo);

	    //
	    // Replace non-CORBA exceptions with CORBA::UNKNOWN
	    //
	    CORBA_UNKNOWN ex;

	    CORBA_ULong count = offsetOut;
	    OBMarshalCount(ex, count);
		
	    buf.alloc(count);
	    CORBA_Octet* oct = buf.data + offsetOut;
	    OBMarshal(ex, oct);
		
	    repH.reply_status = GIOP_SYSTEM_EXCEPTION;
	}
    
	//
        // Create reply message
	//
        GIOP_MessageHeader hdr =
        {
	    { 'G', 'I', 'O', 'P' }, { 1, 0 },
	    OBEndian, GIOP_Reply, buf.len - 12
	};

        CORBA_Octet* oct = buf.data;
        OBMarshal(hdr, oct);
        OBMarshal(repH, oct);

	//
	// Convert simple buffer to OCI buffer
	// Add the OCI buffer
	//
	OCI_Buffer_var b = new OCI_Buffer;
	b -> buf().consume(buf);
	add(b._retn());
    }
    else
    {
	//
	// Execute request, using the BOA
	//
	CORBA_BOA_ptr boa = CORBA_BOA::_OB_instance_nodup();

	try
	{
            //
            // We set the current OCI transport. We need to save the
            // transport so that we can restore the current transport
            // after the call.
            //
            OCI_Current_impl* current = OCI_Current_impl::_OB_impl();
            OCI_TransportInfo_var save =
                current -> _OB_setOCITransportInfo(transportInfo);
            
            try
            {
		OB_DEC_TRY_FIX
                //
                // Since I don't send a response, I'm not interested
                // in the status
                //
                boa -> _OB_dispatch(reqH.object_key, reqH.operation,
                                    buf, swap, offsetIn, 1);
            }
            catch(...)
            {
                //
                // We need to save the return value of to a _var type to
                // ensure that the reference is released.
                //
                OCI_TransportInfo_var dummy =
		    current -> _OB_setOCITransportInfo(save);
                throw;
            }

	    //
	    // We need to save the return value of to a _var type to
	    // ensure that the reference is released.
	    //
            OCI_TransportInfo_var dummy =
                current -> _OB_setOCITransportInfo(save);
        }
	catch(const CORBA_SystemException&)
	{
	    //
	    // Since I don't send a response, I'm not interested in
	    // exceptions either
	    //
	}
        catch(const CORBA_UserException&)
	{
	    //
	    // Print a warning message
	    //
	    displayWarning(
		"Servant method raised an undeclared CORBA user exception",
		reqH.operation,
		transportInfo);

	    //
	    // Since I don't send a response, I'm not interested in
	    // exceptions either
	    //
	}
	catch(...)
	{
	    //
	    // Print a warning message
	    //
	    displayWarning(
		"Servant method raised a non-CORBA exception",
		reqH.operation,
		transportInfo);

	    //
	    // Since I don't send a response, I'm not interested in
	    // exceptions either
	    //
	}
    }
}

// ----------------------------------------------------------------------
// OBGIOPServerWorker constructor and destructor
// ----------------------------------------------------------------------

OBGIOPServerWorker::OBGIOPServerWorker(OBGIOPServerStarter_ptr starter,
				       OCI_Transport_ptr transport)
    : starter_(OBGIOPServerStarter::_duplicate(starter)),
      transport_(OCI_Transport::_duplicate(transport))
{
    //cout << "OBGIOPServerWorker" << endl;
}

OBGIOPServerWorker::~OBGIOPServerWorker()
{
    //cout << "~OBGIOPServerWorker" << endl;
}

// ----------------------------------------------------------------------
// OBGIOPServerWorker public member implementations
// ----------------------------------------------------------------------

/* No public members */

// ----------------------------------------------------------------------
// OBGIOPServerWorkerBlocking constructor and destructor
// ----------------------------------------------------------------------

OBGIOPServerWorkerBlocking::OBGIOPServerWorkerBlocking(
    OBGIOPServerStarterBlocking_ptr starter, OCI_Transport_ptr transport)
    : OBGIOPServerWorker(starter, transport)
{
}

OBGIOPServerWorkerBlocking::~OBGIOPServerWorkerBlocking()
{
}

// ----------------------------------------------------------------------
// OBGIOPServerWorkerBlocking public member implementations
// ----------------------------------------------------------------------

void
OBGIOPServerWorkerBlocking::destroy()
{
    //
    // Add a CloseConnection message
    //
    addCloseConnection();

    //
    // Send all remaining messages, ignoring any errors
    //
    sendRemainingMessages();

    //
    // Close the transport
    //
    closeTransport();
}

void
OBGIOPServerWorkerBlocking::getRequest()
{
    try
    {
	//
	// Setup the incoming message buffer
	//
	assert(CORBA_is_nil(buf_));
	setupBuffer();

	//
	// Receive header, blocking, detect connection loss
	//
	if(!transport_ -> receive_detect(buf_, true))
	{
	    starter_ -> destroyWorker(this);
	    return; // Orderly shutdown
	}
	assert(buf_ -> is_full());

	//
	// Header is complete
	//
	extractHeader();

	if(!buf_ -> is_full())
	{
	    //
	    // Receive body, blocking, detect connection loss
	    //
	    if(!transport_ -> receive_detect(buf_, true))
	    {
		starter_ -> destroyWorker(this);
		return; // Orderly shutdown
	    }
	    assert(buf_ -> is_full());
	}

	//
	// Execute the message
	//
	execute();
    }
    catch(const CORBA_SystemException&)
    {
	//
	// Destroy the worker and rethrow
	//
	starter_ -> destroyWorker(this);
	throw; // In blocking mode, I want exceptions to be propagated
    }
}

void
OBGIOPServerWorkerBlocking::sendReply()
{
    try
    {
	//
	// Try to send all messages
	//
	assert(outSeq_.length() <= 1);
	for(CORBA_ULong i = 0 ; i < outSeq_.length() ; i++)
	{
	    //
	    // Send buffer, blocking, detect connection loss
	    //
	    OCI_Buffer_ptr buf = outSeq_[i];
	    if(!transport_ -> send_detect(buf, true))
	    {
		starter_ -> destroyWorker(this);
		return; // Orderly shutdown
	    }
	    assert(buf -> is_full());
	}

	//
	// Remove all messages
	//
	outSeq_.length(0);
    }
    catch(const CORBA_SystemException&)
    {
	//
	// Destroy the worker and rethrow
	//
	starter_ -> destroyWorker(this);
	throw; // In blocking mode, I want exceptions to be propagated
    }
}

// ----------------------------------------------------------------------
// OBGIOPServerWorkerReactive private and protected member implementations
// ----------------------------------------------------------------------

void
OBGIOPServerWorkerReactive::add(OCI_Buffer_ptr buf)
{
    if(!destroy_) // Don't add if already destroyed
    {
	OBGIOPServerWorker::add(buf);
	
	//
	// Since there is now data to write, register as read and
	// write event handler. This is only necessary if the message
	// added is the first outgoing message in the outgoing message
	// buffer (since if there is more then one message in the
	// buffers, the event handler is already registered).
	//
	if(outSeq_.length() == 1)
	{
	    OCI_Handle handle = transport_ -> handle();
	    if(handle >= 0) // Don't register if already closed
		OBReactor::instance() ->
		    registerHandler(this, OBEventRead | OBEventWrite, handle);
	}
    }
}

// ----------------------------------------------------------------------
// OBGIOPServerWorkerReactive constructor and destructor
// ----------------------------------------------------------------------

OBGIOPServerWorkerReactive::OBGIOPServerWorkerReactive(
    OBGIOPServerStarterReactive_ptr starter, OCI_Transport_ptr transport)
    : OBGIOPServerWorker(starter, transport), destroy_(false)
{
    //
    // Register as read event handler
    //
    OCI_Handle handle = transport_ -> handle();
    assert(handle >= 0);
    OBReactor::instance() -> registerHandler(this, OBEventRead, handle);
}

OBGIOPServerWorkerReactive::~OBGIOPServerWorkerReactive()
{
    assert(destroy_);
}

// ----------------------------------------------------------------------
// OBGIOPServerWorkerReactive public member implementations
// ----------------------------------------------------------------------

void
OBGIOPServerWorkerReactive::destroy()
{
    //
    // WARNING: The reactive worker is not thread safe!
    //

    //
    // Don't destroy twice
    //
    if(destroy_)
	return;

    //
    // Add a CloseConnection message (must be done before destroy_ is
    // set to true)
    //
    addCloseConnection();

    //
    // Send all remaining messages, ignoring any errors
    //
    sendRemainingMessages();

    //
    // Set the destroy flag
    //
    destroy_ = true;
    
    //
    // Unregister as event handler
    //
    OBReactor::instance() -> unregisterHandler(this);

    //
    // Close the transport
    //
    closeTransport();
}

void
OBGIOPServerWorkerReactive::handleEvent(OBMask mask)
{
    //
    // Make sure that this worker cannot be released as long as this
    // operation is executed
    //
    OBGIOPServerWorkerReactive_var worker =
	OBGIOPServerWorkerReactive::_duplicate(this);

    try
    {
	//
	// Handle read events
	//
	if(mask & OBEventRead)
	{
	    //
	    // Setup the incoming message buffer if not already done
	    //
	    if(CORBA_is_nil(buf_))
		setupBuffer();
	    
	    if(buf_ -> pos() < 12)
	    {
		//
		// Receive header, non-blocking, detect connection loss
		//
		if(!transport_ -> receive_detect(buf_, false))
		{
		    starter_ -> destroyWorker(this);
		    return; // Orderly shutdown
		}
		
		//
		// Header complete?
		//
		if(buf_ -> is_full())
		    extractHeader();
	    }
	    
	    if(buf_ -> pos() >= 12)
	    {
		if(!buf_ -> is_full())
		{
		    //
		    // Receive body, non-blocking, detect connection loss
		    //
		    if(!transport_ -> receive_detect(buf_, false))
		    {
			starter_ -> destroyWorker(this);
			return; // Orderly shutdown
		    }
		}
		
		//
		// Execute message if buffer is full
		//
		if(buf_ -> is_full())
		    execute();
	    }
	}
	
	//
	// Handle write events
	//
	if((mask & OBEventWrite) && outSeq_.length() > 0)
	{
	    //
	    // Send buffer, non-blocking, detect connection loss
	    //
	    OCI_Buffer_ptr buf = outSeq_[0];
	    if(!transport_ -> send_detect(buf, false))
	    {
		starter_ -> destroyWorker(this);
		return; // Orderly shutdown
	    }
	    
	    if(buf -> is_full())
	    {
		//
		// Message was sent completely, remove it
		//
		for(CORBA_ULong i = 1 ; i < outSeq_.length() ; i++)
		    outSeq_[i - 1] = outSeq_[i];
		outSeq_.length(outSeq_.length() - 1);
		
		if(outSeq_.length() == 0)
		{
		    //
		    // Register as read event handler only, since
		    // there is nothing to write anymore
		    //
		    OCI_Handle handle = transport_ -> handle();
		    if(handle >= 0) // Don't register if already closed
			OBReactor::instance()
			    -> registerHandler(this, OBEventRead, handle);
		}
	    }
	}
    }
    catch(const CORBA_SystemException&)
    {
	//
	// Destroy the worker and rethrow
	//
	starter_ -> destroyWorker(this);
	//throw; // I don't want exceptions to be propagated
    }
}

#ifdef HAVE_JTC

void
OBGIOPServerWorkerThreaded::add(OCI_Buffer_ptr buf)
{
    JTCSynchronized synchronized(*this);

    if(!destroy_) // Don't add if already destroyed
    {
	OBGIOPServerWorker::add(buf);
	
	//
	// Notify the sender thread. This is only necessary if this is
	// the first outgoing message, since if there is more then one
	// message in the buffers, threads are already notified.
	//
	if(outSeq_.length() == 1)
	    notifyAll();
    }
}

OCI_Buffer_ptr
OBGIOPServerWorkerThreaded::remove()
{
    JTCSynchronized synchronized(*this);

    //
    // Wait for a message
    //
    while(!destroy_ && outSeq_.length() == 0)
    {
	try
	{
	    wait();
	}
	catch(const JTCInterruptedException&)
	{
	}
    }

    //
    // Return nil buffer if destroy_ was set
    //
    if(destroy_)
	return OCI_Buffer::_nil();

    //
    // Remove and return first message from message buffers
    //
    assert(outSeq_.length() > 0);
    OCI_Buffer_var buf = outSeq_[0];
    
    for(CORBA_ULong i = 1 ; i < outSeq_.length() ; i++)
	outSeq_[i - 1] = outSeq_[i];
    outSeq_.length(outSeq_.length() - 1);
    
    return buf._retn();
}

void
OBGIOPServerWorkerThreaded::dispatchRequest(OBBuffer& buf, bool swap)
{
    switch(CORBA_BOA::conc_model())
    {
    case CORBA_BOA::ConcModelBlocking:
    case CORBA_BOA::ConcModelReactive:
	assert(false);
	break;

    case CORBA_BOA::ConcModelThreaded:
    {
	JTCSynchronized synchronized(singleThreadedMutex_);
	OBGIOPServerWorker::dispatchRequest(buf, swap);
	break;
    }
    
    case CORBA_BOA::ConcModelThreadPerClient:
	OBGIOPServerWorker::dispatchRequest(buf, swap);
	break;
	
    case CORBA_BOA::ConcModelThreadPerRequest:
    {
	(new Dispatcher(this, buf, swap)) -> start();
	break;
    }

    case CORBA_BOA::ConcModelThreadPool:
    {
	ThreadPool* pool = ThreadPool::instance();
	pool -> add(this, buf, swap);
	break;
    }
    }
}

// ----------------------------------------------------------------------
// OBGIOPServerWorkerThreaded constructor and destructor
// ----------------------------------------------------------------------

OBGIOPServerWorkerThreaded::OBGIOPServerWorkerThreaded(
    OBGIOPServerStarterThreaded_ptr starter, OCI_Transport_ptr transport)
    : OBGIOPServerWorker(starter, transport), destroy_(false)
{
    try
    {
	OB_DEC_TRY_FIX

	//
	// Start sender and receiver thread
	//
	senderThread_ = new SenderThread(this);
	senderThread_ -> start();
	receiverThread_ = new ReceiverThread(this);
	receiverThread_ -> start();
    }
    catch(...)
    {
	_OB_setRef(0);
	throw;
    }
}

OBGIOPServerWorkerThreaded::~OBGIOPServerWorkerThreaded()
{
    assert(destroy_);
}

// ----------------------------------------------------------------------
// OBGIOPServerWorkerThreaded public member implementations
// ----------------------------------------------------------------------

void
OBGIOPServerWorkerThreaded::destroy()
{
    JTCSynchronized synchronized(*this);

    //
    // Don't destroy twice
    //
    if(destroy_)
	return;

    //
    // Add a CloseConnection message (must be done before destroy_ is
    // set to true)
    //
    addCloseConnection();

    //
    // Set the destroy flag
    //
    destroy_ = true;
    
    //
    // Notify the sender and the receiver thread
    //
    notifyAll();
    
    //
    // Wait for the sender thread to terminate
    //
    if(senderThread_ &&
       senderThread_ -> getId() != JTCThread::currentThread() -> getId())
    {
	while(senderThread_)
	{
	    try
	    {
		wait();
	    }
	    catch(const JTCInterruptedException&)
	    {
	    }
	}
    }

    //
    // Unblock threads blocking in the send() and recv() operations
    //
    transport_ -> unblock_threads(CORBA_FALSE);
    
    //
    // Wait for the receiver thread to terminate
    //
    if(receiverThread_ &&
       receiverThread_ -> getId() != JTCThread::currentThread() -> getId())
    {
	while(receiverThread_)
	{
	    try
	    {
		wait();
	    }
	    catch(const JTCInterruptedException&)
	    {
	    }
	}
    }

    //
    // Close the transport
    //
    closeTransport();
}

void
OBGIOPServerWorkerThreaded::senderRun()
{
    while(!destroy_)
    {
	try
	{
	    //
	    // Remove outgoing message from the outgoing message buffers
	    //
	    OCI_Buffer_var buf = remove();

	    //
	    // Check if thread should be stopped and destroyed
	    //
	    if(CORBA_is_nil(buf))
		break;

	    //
	    // Send buffer, blocking, detect connection loss
	    //
	    if(!transport_ -> send_detect(buf, true))
	    {
		//
		// Destroy the worker
		//
	        starter_ -> destroyWorker(this);

		return; // Orderly shutdown
	    }
	    assert(buf -> is_full());
	}
	catch(const CORBA_SystemException&)
	{
	    //
	    // Destroy the worker
	    //
	    starter_ -> destroyWorker(this);

	    return;
	}
    }

    //
    // Send all remaining messages, ignoring any errors
    //
    sendRemainingMessages();
}

void
OBGIOPServerWorkerThreaded::receiverRun()
{
    while(!destroy_)
    {
	try
	{
	    //
	    // Setup the incoming message buffer
	    //
	    assert(CORBA_is_nil(buf_));
	    setupBuffer();
	    
	    //
	    // Receive header, blocking, detect connection loss
	    //
	    if(!transport_ -> receive_detect(buf_, true))
	    {
		//
		// Destroy the worker
		//
		starter_ -> destroyWorker(this);

		return; // Orderly shutdown
	    }
            assert(buf_ -> is_full());

	    //
	    // Check if thread should be stopped and destroyed
            //
	    if(destroy_)
		break;

	    //
	    // Header is complete
	    //
	    extractHeader();
	    
	    if(!buf_ -> is_full())
	    {
		//
		// Receive body, blocking, detect connection loss
		//
		if(!transport_ -> receive_detect(buf_, true))
		{
		    //
		    // Destroy the worker
		    //
		    starter_ -> destroyWorker(this);

		    return; // Orderly shutdown
		}
		assert(buf_ -> is_full());
	    }
	    
	    //
	    // Execute the message
	    //
	    if(!destroy_)
		execute();
	}
	catch(const CORBA_SystemException&)
	{
	    //
	    // Destroy the worker
	    //
	    starter_ -> destroyWorker(this);

	    return;
	}
    }
}

// ----------------------------------------------------------------------
// OBGIOPServerWorkerThreaded::SenderThread public member implementations
// ----------------------------------------------------------------------

void
OBGIOPServerWorkerThreaded::SenderThread::run()
{
    worker_ -> senderRun();
    
    {
	JTCSynchronized sync(*(worker_.in()));
	worker_ -> senderThread_ = JTCThreadHandle();
	worker_ -> notifyAll();
    }
    
    // Break cyclic object dependency
    worker_ = OBGIOPServerWorkerThreaded::_nil();
}

// ----------------------------------------------------------------------
// OBGIOPServerWorkerThreaded::ReceiverThread public member implementations
// ----------------------------------------------------------------------

void
OBGIOPServerWorkerThreaded::ReceiverThread::run()
{
    worker_ -> receiverRun();
    
    {
	JTCSynchronized sync(*(worker_.in()));
	worker_ -> receiverThread_ = JTCThreadHandle();
	worker_ -> notifyAll();
    }
    
    // Break cyclic object dependency
    worker_ = OBGIOPServerWorkerThreaded::_nil();
}

// ----------------------------------------------------------------------
// OBGIOPServerWorkerThreaded::ThreadPool private and protected member
// implementations
// ----------------------------------------------------------------------

void
OBGIOPServerWorkerThreaded::ThreadPool::get(
    OBGIOPServerWorkerThreaded::ThreadPool::Request& req)
{
    JTCSynchronized synchronized(*this);
    
    while(!destroy_ && !head_)
    {
	try
	{
	    wait();
	}
	catch(const JTCInterruptedException&)
	{
	}
    }

    if(destroy_)
    {
	req.worker = OBGIOPServerWorkerThreaded::_nil();
	return;
    }

    req.worker = head_ -> worker;
    req.buf.consume(head_ -> buf);
    req.swap = head_ -> swap;
    req.next = 0;
    
    Request* next = head_ -> next;
    delete head_;
    head_ = next;
}

// ----------------------------------------------------------------------
// OBGIOPServerWorkerThreaded::ThreadPool constructor and destructor
// ----------------------------------------------------------------------

OBGIOPServerWorkerThreaded::ThreadPool::ThreadPool()
    : destroy_(false), head_(0)
{
    num_ = CORBA_BOA::conc_model_thread_pool();
    handles_ = new JTCThreadHandle[num_];
    for(CORBA_ULong i = 0 ; i < num_ ; i++)
    {
	handles_[i] = new Dispatcher(this);
	handles_[i] -> start();
    }
}

OBGIOPServerWorkerThreaded::ThreadPool::~ThreadPool()
{
    assert(destroy_);
    
    delete [] handles_;
    
    while(head_)
    {
	Request* next = head_ -> next;
	delete head_;
	head_ = next;
    }
}

// ----------------------------------------------------------------------
// OBGIOPServerWorkerThreaded::ThreadPool public
// ----------------------------------------------------------------------

OBGIOPServerWorkerThreaded::ThreadPool*
OBGIOPServerWorkerThreaded::ThreadPool::instance()
{
    if(!instance_) // Double-checked locking
    {
	JTCSynchronized synchronized(instanceMutex_);
	
        if(!instance_)
	    instance_ = new ThreadPool();
    }
    
    return instance_;
}

void
OBGIOPServerWorkerThreaded::ThreadPool::destroyInstance()
{
    if(instance_) // Double-checked locking
    {
	JTCSynchronized synchronized(instanceMutex_);
	
        if(instance_)
	{
	    instance_ -> destroy();
	    instance_ = 0;
	}
    }
}

// TODO: This is a hack!
void
OBDestroyThreadPool()
{
    OBGIOPServerWorkerThreaded::ThreadPool::destroyInstance();
} 

void
OBGIOPServerWorkerThreaded::ThreadPool::destroy()
{
    {
	JTCSynchronized synchronized(*this);
	destroy_ = true;
	notifyAll();
    }

    for(CORBA_ULong i = 0 ; i < num_ ; i++)
    {
	while(handles_[i] -> isAlive())
	{
	    try
	    {
		handles_[i] -> join();
	    }
	    catch(const JTCInterruptedException&)
	    {
	    }
	}
    }

    delete this;
}

void
OBGIOPServerWorkerThreaded::ThreadPool::add(
    OBGIOPServerWorkerThreaded_ptr w, OBBuffer& b, bool s)
{
    JTCSynchronized synchronized(*this);
    
    Request** p = &head_;
    while(*p)
	p = &((*p) -> next);
    
    *p = new Request;
    (*p) -> worker = OBGIOPServerWorkerThreaded::_duplicate(w);
    (*p) -> buf.consume(b);
    (*p) -> swap = s;
    (*p) -> next = 0;
    
    notify();
}

#endif
