// **********************************************************************
//
// 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/IOP.h>
#include <OB/Object.h>
#include <OB/GIOP.h>
#include <OB/OCI.h>
#include <OB/Reactor.h>

#include <GIOPClient.h>
#include <GIOPClientWorker.h>

#ifdef WIN32
#   include <sys/timeb.h>
#endif

//
// Next request ID
//
CORBA_ULong OBGIOPClientWorker::nextRequestId_ = 0;

//
// Discarded request IDs
//
OBFixSeq< CORBA_ULong > OBGIOPClientWorker::discReqIdSeq_;

#ifdef HAVE_JTC
//
// Mutex for request ID handling
//
JTCMutex OBGIOPClientWorker::requestIdMutex_;
#endif

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

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

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

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

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

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

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

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

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

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

#ifndef HAVE_NO_EXPLICIT_TEMPLATES
template class OBObjVar< OBGIOPClientWorker >;
template class OBObjForSeq< OBGIOPClientWorker >;
//template class OBObjVar< OBGIOPClientWorkerBlocking >;
//template class OBObjForSeq< OBGIOPClientWorkerBlocking >;
template class OBObjVar< OBGIOPClientWorkerReactive >;
template class OBObjForSeq< OBGIOPClientWorkerReactive >;
#ifdef HAVE_JTC
//template class OBObjVar< OBGIOPClientWorkerThreaded >;
//template class OBObjForSeq< OBGIOPClientWorkerThreaded >;
#endif
#else
#ifdef HAVE_PRAGMA_DEFINE
#pragma define(OBObjVar< OBGIOPClientWorker >)
#pragma define(OBObjForSeq< OBGIOPClientWorker >)
//#pragma define(OBObjVar< OBGIOPClientWorkerBlocking >)
//#pragma define(OBObjForSeq< OBGIOPClientWorkerBlocking >)
#pragma define(OBObjVar< OBGIOPClientWorkerReactive >)
#pragma define(OBObjForSeq< OBGIOPClientWorkerReactive >)
#ifdef HAVE_JTC
//#pragma define(OBObjVar< OBGIOPClientWorkerThreaded >)
//#pragma define(OBObjForSeq< OBGIOPClientWorkerThreaded >)
#endif
#endif
#endif

// ----------------------------------------------------------------------
// OBGIOPClientWorker private and protected member implementations
// ----------------------------------------------------------------------

void
OBGIOPClientWorker::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);
}

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

void
OBGIOPClientWorker::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_MAYBE);
    }
    
    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);
}

void
OBGIOPClientWorker::add(OCI_Buffer_ptr buf)
{
    //
    // Add new message to the message buffers
    //
    outSeq_.length(outSeq_.length() + 1);
    outSeq_[outSeq_.length() - 1] = OCI_Buffer::_duplicate(buf);
}

void
OBGIOPClientWorker::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
OBGIOPClientWorker::closeTransport()
{
    //
    // `discard' parameter must be set to TRUE, as it is not necessary
    // to reliably transmit outstanding messages.
    //
    transport_ -> close(CORBA_TRUE);
}

void
OBGIOPClientWorker::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_MAYBE);
    }
    
    switch(type)
    {
    case GIOP_Reply:
    {
	const CORBA_Octet* coct = buf.data + 12;
	GIOP_ReplyHeader repH;
	OBUnmarshal(repH, coct, swap);

	//
	// Check if request has been discarded
	//
	bool discard = false;
	{
#ifdef HAVE_JTC
	    JTCSynchronized synchronized(requestIdMutex_);
#endif
	    for(CORBA_ULong i = 0 ; i < discReqIdSeq_.length() ; i++)
		if(discReqIdSeq_[i] == repH.request_id)
		{
		    discard = true;
		    discReqIdSeq_.remove(i);
		    break;
		}
	}
	
	if(!discard)
	{
	    //
	    // Save the message
	    //
	    Reply* rep = new Reply;
	    rep -> reqId = repH.request_id;
	    rep -> buf.consume(buf);
	    rep -> swap = swap;
	    rep -> status = repH.reply_status;
	    rep -> offset = coct - rep -> buf.data;
	    rep -> next = head_;
	    head_ = rep;
	}
    }
    
    break;
    
    case GIOP_Request:
    case GIOP_LocateRequest:
    case GIOP_CancelRequest:
	
	//
	// These messages may not be sent to a client
	//
	addMessageError();
	throw CORBA_COMM_FAILURE("", OBMinorWrongMessage,
				 CORBA_COMPLETED_MAYBE);
	
    case GIOP_LocateReply:
	
	//
	// Since I am not sending LocatRequest messages,
	// LocateReply may not be sent to this client
	//
	addMessageError();
	throw CORBA_COMM_FAILURE("", OBMinorWrongMessage,
				 CORBA_COMPLETED_MAYBE);
	
    case GIOP_CloseConnection:
	
	//
	// Notify connection closing by throwing a TRANSIENT exception
	//
	throw CORBA_TRANSIENT();
	
    case GIOP_MessageError:
	
	//
	// How are message error messages handled? I just send an
	// error message back.
	//
	addMessageError();
	throw CORBA_COMM_FAILURE("", OBMinorMessageError,
				 CORBA_COMPLETED_MAYBE);
    }
}

CORBA_ULong
OBGIOPClientWorker::findReply(CORBA_ULong reqId, OBBuffer& buf,
			      bool& swap, GIOP_ReplyStatusType& status)
{
    //
    // Check buffered reply messages
    //
    Reply** p = &head_;
    while(*p)
    {
	Reply* rep = *p;

	if(rep -> reqId == reqId)
	{
	    //
	    // Reply found
	    //
	    buf.consume(rep -> buf);
	    swap = rep -> swap;
	    status = rep -> status;
	    CORBA_ULong offset = rep -> offset;

	    //
	    // Remove reply from buffer
	    //
	    *p = rep -> next;
	    delete rep;
	    
	    return offset;
	}
	else
	{
	    p = &(rep -> next);
	}
    }
    
    //
    // Nothing found in buffer
    //
    return 0;
}

CORBA_Boolean
OBGIOPClientWorker::checkReply(CORBA_ULong reqId)
{
    //
    // Check buffered reply messages
    //
    Reply* rep = head_;
    while(rep)
    {
	if(rep -> reqId == reqId)
	    return true;

	rep = rep -> next;
    }
    
    //
    // Nothing found in buffer
    //
    return false;
}

// ----------------------------------------------------------------------
// OBGIOPClientWorker constructor and destructor
// ----------------------------------------------------------------------

OBGIOPClientWorker::OBGIOPClientWorker(OBGIOPClient_ptr client,
				       OCI_Transport_ptr transport)
    : client_(OBGIOPClient::_duplicate(client)),
      transport_(OCI_Transport::_duplicate(transport)), head_(0)
{
    //cout << "OBGIOPClientWorker" << endl;
}

OBGIOPClientWorker::~OBGIOPClientWorker()
{
    //cout << "~OBGIOPClientWorker" << endl;

    //
    // Remove all buffered reply messages
    //
    while(head_)
    {
	Reply* next = head_ -> next;
	delete head_;
	head_ = next;
    }
}

// ----------------------------------------------------------------------
// OBGIOPClientWorker public member implementations
// ----------------------------------------------------------------------

OCI_Transport_ptr
OBGIOPClientWorker::transport()
{
    return OCI_Transport::_duplicate(transport_);
}

// ----------------------------------------------------------------------
// OBGIOPClientWorkerBlocking constructor and destructor
// ----------------------------------------------------------------------

OBGIOPClientWorkerBlocking::OBGIOPClientWorkerBlocking(OBGIOPClient_ptr client,
						       OCI_Transport_ptr
						       transport)
    : OBGIOPClientWorker(client, transport)
{
}

OBGIOPClientWorkerBlocking::~OBGIOPClientWorkerBlocking()
{
}

// ----------------------------------------------------------------------
// OBGIOPClientWorkerBlocking public member implementations
// ----------------------------------------------------------------------

void
OBGIOPClientWorkerBlocking::destroy()
{
    //
    // Send all remaining messages, ignoring any errors
    //
    sendRemainingMessages();

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

CORBA_ULong
OBGIOPClientWorkerBlocking::sendReceiveTimeout(CORBA_ULong reqId,
					       OBBuffer& buf, bool& swap,
					       GIOP_ReplyStatusType& status,
					       CORBA_Long t)
{
    OB_SYNCHRONIZED

    //
    // Convert simple buffer to OCI buffer
    // Add the OCI buffer
    //
    OCI_Buffer_var b = new OCI_Buffer;
    b -> buf().consume(buf);
    add(b);

    //
    // Try to send all messages
    //
    try
    {
	if(t < 0)
	{
	    for(CORBA_ULong i = 0 ; i < outSeq_.length() ; i++)
	    {
		//
		// Send buffer, blocking
		//
		OCI_Buffer_ptr buf = outSeq_[i];
		transport_ -> send(buf, true);
		assert(buf -> is_full());
	    }

	    //
	    // Remove all messages
	    //
	    outSeq_.length(0);
	}
	else
	{
	    //
	    // Get current time
	    //
#ifdef WIN32
	    struct _timeb tb1;
	    _ftime(&tb1);
#else
	    timeval tv1;
	    gettimeofday(&tv1, 0);
#endif
	    CORBA_Long millis = 0;

	    while(outSeq_.length() > 0)
	    {
		//
		// Send buffer, with timeout
		//
		OCI_Buffer_ptr buf = outSeq_[0];
		CORBA_ULong oldPos = buf -> pos();
		transport_ -> send_timeout(buf, (CORBA_ULong)t);
		if(oldPos == buf -> pos())
		    return 0; // Timeout

		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);
		}

		//
		// Calculate time needed
		//
#ifdef WIN32
		struct _timeb tb2;
		_ftime(&tb2);
		millis = (tb2.time - tb1.time) * 1000;
		millis += tb2.millitm;
		millis -= tb1.millitm;
#else
		timeval tv2;
		gettimeofday(&tv2, 0);
		millis = (tv2.tv_sec - tv1.tv_sec) * 1000;
		millis += tv2.tv_usec / 1000;
		millis -= tv1.tv_usec / 1000;
#endif
		//
		// Check if time limit has been exceeded
		//
		if(millis >= t)
		    return 0; // Timeout
	    }

	    //
	    // Calculate new timeout
	    //
	    t -= millis;
	    if(t <= 0)
		return 0; // Timeout
	}
    }
    catch(const CORBA_SystemException&)
    {
	//
	// Restore the buffer
	//
	buf.consume(b -> buf());

	//
	// Destroy the worker and rethrow
	//
	client_ -> destroyWorker(this);
	throw;
    }
    
    //
    // Return the reply
    //
    // Upon communication failure, the buffer may not be restored, in
    // order to guarantee "at most once" semantics
    //
    return receiveTimeout(reqId, buf, swap, status, t);
}

CORBA_ULong
OBGIOPClientWorkerBlocking::receiveTimeout(CORBA_ULong reqId, OBBuffer& buf,
					   bool& swap,
					   GIOP_ReplyStatusType& status,
					   CORBA_Long t)
{
    OB_SYNCHRONIZED

    CORBA_ULong offset = 0;

    //
    // Try to receive the reply
    //
    try
    {
	if(t < 0)
	{
	    while((offset = findReply(reqId, buf, swap, status)) == 0)
	    {
		//
		// Setup the incoming message buffer
		//
		//assert(CORBA_is_nil(buf_)); // This assert seems to be wrong
		if(CORBA_is_nil(buf_))
		    setupBuffer();

		//
		// Receive header, blocking
		//
		transport_ -> receive(buf_, true);
		assert(buf_ -> is_full());

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

		if(!buf_ -> is_full())
		{
		    //
		    // Receive body, blocking
		    //
		    transport_ -> receive(buf_, true);
		    assert(buf_ -> is_full());
		}

		//
		// Execute the message
		//
		execute();
	    }
	}
	else
	{
	    //
	    // Get current time
	    //
#ifdef WIN32
	    struct _timeb tb1;
	    _ftime(&tb1);
#else
	    timeval tv1;
	    gettimeofday(&tv1, 0);
#endif
	    CORBA_Long millis = 0;

	    while((offset = findReply(reqId, buf, swap, status)) == 0)
	    {
		//
		// Setup the incoming message buffer if not already done
		//
		if(CORBA_is_nil(buf_))
		    setupBuffer();

		if(buf_ -> pos() < 12)
		{
		    //
		    // Receive header, with timeout
		    //
		    CORBA_ULong oldPos = buf_ -> pos();
		    transport_ -> receive_timeout(buf_, t);
		    if(oldPos == buf_ -> pos())
			return 0; // Timeout

		    //
		    // Header complete?
		    //
		    if(buf_ -> is_full())
			extractHeader();
		}

		if(buf_ -> pos() >= 12)
		{
		    if(!buf_ -> is_full())
		    {
			//
			// Receive body, with timeout
			//
			CORBA_ULong oldPos = buf_ -> pos();
			transport_ -> receive_timeout(buf_, t);
			if(oldPos == buf_ -> pos())
			    return 0; // Timeout
		    }

		    //
		    // Execute message if buffer is full
		    //
		    if(buf_ -> is_full())
			execute();
		}

		//
		// Calculate time needed
		//
#ifdef WIN32
		struct _timeb tb2;
		_ftime(&tb2);
		millis = (tb2.time - tb1.time) * 1000;
		millis += tb2.millitm;
		millis -= tb1.millitm;
#else
		timeval tv2;
		gettimeofday(&tv2, 0);
		millis = (tv2.tv_sec - tv1.tv_sec) * 1000;
		millis += tv2.tv_usec / 1000;
		millis -= tv1.tv_usec / 1000;
#endif
		//
		// Check if time limit has been exceeded
		//
		if(millis >= t)
		    return 0; // Timeout
	    }
	}
    }
    catch(const CORBA_SystemException&)
    {
	//
	// Destroy the worker and rethrow
	//
	client_ -> destroyWorker(this);
	throw;
    }

    assert(offset);
    return offset;
}

void
OBGIOPClientWorkerBlocking::sendNonBlocking(OBBuffer& buf)
{
    OB_SYNCHRONIZED

    //
    // Convert simple buffer to OCI buffer
    // Add the OCI buffer
    //
    OCI_Buffer_var b = new OCI_Buffer;
    b -> buf().consume(buf);
    add(b);

    //
    // Try to send all messages
    //
    try
    {
	while(outSeq_.length() > 0)
	{
	    //
	    // Send buffer, non-blocking
	    //
	    OCI_Buffer_ptr buf = outSeq_[0];
	    CORBA_ULong oldPos = buf -> pos();
	    transport_ -> send(buf, false);
	    if(oldPos == buf -> pos())
		break; // Nothing was sent

	    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);
	    }
	}
    }
    catch(const CORBA_SystemException&)
    {
	//
	// Restore the buffer
	//
	buf.consume(b -> buf());

	//
	// Destroy the worker and rethrow
	//
	client_ -> destroyWorker(this);
	throw;
    }
}

// ---------------------------------------------------------------------
// OBGIOPClientWorkerReactive private and protected member implementations
// ----------------------------------------------------------------------

void
OBGIOPClientWorkerReactive::add(OCI_Buffer_ptr buf)
{
    //OB_SYNCHRONIZED
    // Synchronization can be left out for performance reasons, since
    // this operation is only called from operations which are already
    // synchronized.

    if(!destroy_) // Don't add if already destroyed
    {
	OBGIOPClientWorker::add(buf);
	
	//
	// Since there is now data to write, register as read and write
	// event handler. This is only necessary if this is the first
	// outgoing message, 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();
	    assert(handle >= 0);
	    OBReactor::instance() ->
		registerHandler(this, OBEventRead | OBEventWrite, handle);
	}
    }
}

// ----------------------------------------------------------------------
// OBGIOPClientWorkerReactive constructor and destructor
// ----------------------------------------------------------------------

OBGIOPClientWorkerReactive::OBGIOPClientWorkerReactive(OBGIOPClient_ptr client,
						       OCI_Transport_ptr
						       transport)
    : OBGIOPClientWorker(client, transport), destroy_(false), exception_(0)
{
    //
    // Register as read event handler
    //
    OCI_Handle handle = transport_ -> handle();
    assert(handle >= 0);
    OBReactor::instance() -> registerHandler(this, OBEventRead, handle);
}

OBGIOPClientWorkerReactive::~OBGIOPClientWorkerReactive()
{
    assert(destroy_);
    delete exception_;
}

// ----------------------------------------------------------------------
// OBGIOPClientWorkerReactive public member implementations
// ----------------------------------------------------------------------

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

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

    //
    // 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();
}

CORBA_ULong
OBGIOPClientWorkerReactive::sendReceiveTimeout(CORBA_ULong reqId,
					       OBBuffer& buf, bool& swap,
					       GIOP_ReplyStatusType& status,
					       CORBA_Long t)
{
    OB_SYNCHRONIZED

    //
    // Dispatch events using zero-timeout, in order to detect
    // communication failures
    //
    if(!exception_)
    {
	if(t < 0)
	{
	    while(OBReactor::instance() -> dispatchOneEvent(0))
		;
	}
	else
	{
	    //
	    // Get current time
	    //
#ifdef WIN32
	    struct _timeb tb1;
	    _ftime(&tb1);
#else
	    timeval tv1;
	    gettimeofday(&tv1, 0);
#endif
	    CORBA_Long millis = 0;
	    
	    while(OBReactor::instance() -> dispatchOneEvent(0))
	    {
		//
		// Calculate time needed
		//
#ifdef WIN32
		struct _timeb tb2;
		_ftime(&tb2);
		millis = (tb2.time - tb1.time) * 1000;
		millis += tb2.millitm;
		millis -= tb1.millitm;
#else
		timeval tv2;
		gettimeofday(&tv2, 0);
		millis = (tv2.tv_sec - tv1.tv_sec) * 1000;
		millis += tv2.tv_usec / 1000;
		millis -= tv1.tv_usec / 1000;
#endif
		//
		// Check if time limit has been exceeded
		//
		if(millis >= t)
		    return 0; // Timeout
            }
	    
	    //
	    // Calculate new timeout
	    //
	    t -= millis;
	    if(t <= 0)
		return 0; // Timeout
	}
    }

    //
    // Check whether there is an exception
    //
    if(exception_)
    {
	//
	// Restoring the buffer is not necessary here, since it has
	// not been converted into an OCI buffer yet
	//

	//
	// Destroy the worker and raise the exception
	//
	client_ -> destroyWorker(this);
	exception_ -> _raise();
    }

    //
    // Convert simple buffer to OCI buffer
    // Add the OCI buffer
    //
    OCI_Buffer_var b = new OCI_Buffer;
    b -> buf().consume(buf);
    add(b);

    //
    // Return the reply
    //
    // Upon communication failure, the buffer may not be restored, in
    // order to guarantee "at most once" semantics
    //
    return receiveTimeout(reqId, buf, swap, status, t);
}

CORBA_ULong
OBGIOPClientWorkerReactive::receiveTimeout(CORBA_ULong reqId, OBBuffer& buf,
					   bool& swap,
					   GIOP_ReplyStatusType& status,
					   CORBA_Long t)
{
    OB_SYNCHRONIZED

    CORBA_ULong offset = 0;

    if(t < 0)
    {
	//
	// Wait for reply or exception
	//
	while(!exception_ &&
	      (offset = findReply(reqId, buf, swap, status)) == 0)
	{
	    //
	    // Dispatch one event, using the reactor
	    //
	    if(!OBReactor::instance() -> dispatchOneEvent(t))
		assert(false);
	}	
    }
    else
    {
	//
	// Get current time
	//
#ifdef WIN32
	struct _timeb tb1;
	_ftime(&tb1);
#else
	timeval tv1;
	gettimeofday(&tv1, 0);
#endif
	CORBA_Long millis = 0;
	    
	//
	// Wait for reply or exception
	//
	while(!exception_ &&
	      (offset = findReply(reqId, buf, swap, status)) == 0)
	{
	    //
	    // Dispatch one event, using the reactor
	    //
	    if(!OBReactor::instance() -> dispatchOneEvent(t))
		return 0; // Timeout

	    //
	    // Calculate time needed
	    //
#ifdef WIN32
	    struct _timeb tb2;
	    _ftime(&tb2);
	    millis = (tb2.time - tb1.time) * 1000;
	    millis += tb2.millitm;
	    millis -= tb1.millitm;
#else
	    timeval tv2;
	    gettimeofday(&tv2, 0);
	    millis = (tv2.tv_sec - tv1.tv_sec) * 1000;
	    millis += tv2.tv_usec / 1000;
	    millis -= tv1.tv_usec / 1000;
#endif
	    //
	    // Check if time limit has been exceeded
	    //
	    if(millis >= t)
		return 0; // Timeout
	}
    }
    
    //
    // Check whether there is an exception
    //
    if(exception_)
    {
	//
	// Destroy the worker and raise the exception
	//
	client_ -> destroyWorker(this);
	exception_ -> _raise();
    }

    assert(offset);
    return offset;
}

void
OBGIOPClientWorkerReactive::sendNonBlocking(OBBuffer& buf)
{
    OB_SYNCHRONIZED

    //
    // Dispatch events using zero-timeout, in order to detect
    // communication failures
    //
    if(!exception_)
    {
	while(OBReactor::instance() -> dispatchOneEvent(0))
	    ;
    }

    //
    // Check whether there is an exception
    //
    if(exception_)
    {
	//
	// Restoring the buffer is not necessary here, since it has
	// not been converted into an OCI buffer yet
	//

	//
	// Destroy the worker and raise the exception
	//
	client_ -> destroyWorker(this);
	exception_ -> _raise();
    }

    //
    // Convert simple buffer to OCI buffer
    // Add the OCI buffer
    //
    OCI_Buffer_var b = new OCI_Buffer;
    b -> buf().consume(buf);
    add(b);

    //
    // Dispatch events using zero-timeout, giving short messages a
    // chance to be deliviered immediately
    //
    while(OBReactor::instance() -> dispatchOneEvent(0))
	;

    //
    // Check whether there is an exception
    //
    if(exception_)
    {
	//
	// Upon communication failure, the buffer may not be restored,
	// in order to guarantee "at most once" semantics
	//

	//
	// Destroy the worker and raise the exception
	//
	client_ -> destroyWorker(this);
	exception_ -> _raise();
    }

    //
    // Nothing else to do here
    //
}

void
OBGIOPClientWorkerReactive::handleEvent(OBMask mask)
{
    //
    // Make sure that this worker cannot be released as long as this
    // operation is executed.
    //
    // NOTE: This must go before the synchronization!
    //
    OBGIOPClientWorkerReactive_var worker =
	OBGIOPClientWorkerReactive::_duplicate(this);

    OB_SYNCHRONIZED

    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
		//
		transport_ -> receive(buf_, false);
		
		//
		// Header complete?
		//
		if(buf_ -> is_full())
		    extractHeader();
	    }
	    
	    if(buf_ -> pos() >= 12)
	    {
		if(!buf_ -> is_full())
		{
		    //
		    // Receive body, non-blocking
		    //
		    transport_ -> receive(buf_, false);
		}
		
		//
		// Execute message if buffer is full
		//
		if(buf_ -> is_full())
		    execute();
	    }
	}
	
	//
	// Handle write events
	//
	if((mask & OBEventWrite) && outSeq_.length() > 0)
	{
	    //
	    // Send buffer, non-blocking
	    //
	    OCI_Buffer_ptr buf = outSeq_[0];
	    transport_ -> send(buf, false);
	    
	    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();
		    assert(handle >= 0);
		    OBReactor::instance()
			-> registerHandler(this, OBEventRead, handle);
		}
	    }
	}
    }
    catch(const CORBA_SystemException& ex)
    {
	//
	// Buffer the exception if none is set
	//
	if(!exception_)
	{
	    exception_ = CORBA_SystemException::_narrow(ex._OB_clone());
	    assert(exception_);
	}
	
	//
	// The worker may *not* be removed from the client, otherwise
	// the client will not detect the communication failure.
	//
	//client_ -> destroyWorker(this);
	destroy();
    }
}

#ifdef HAVE_JTC

// ----------------------------------------------------------------------
// OBGIOPClientWorkerThreaded private and protected member implementations
// ----------------------------------------------------------------------

void
OBGIOPClientWorkerThreaded::add(OCI_Buffer_ptr buf)
{
    //JTCSynchronized synchronized(*this);
    // Synchronization can be left out for performance reasons, since
    // this operation is only called from operations which are already
    // synchronized.

    if(!destroy_) // Don't add if already destroyed
    {
	OBGIOPClientWorker::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
OBGIOPClientWorkerThreaded::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();
}

// ----------------------------------------------------------------------
// OBGIOPClientWorkerThreaded constructor and destructor
// ----------------------------------------------------------------------

OBGIOPClientWorkerThreaded::OBGIOPClientWorkerThreaded(OBGIOPClient_ptr client,
						       OCI_Transport_ptr
						       transport)
    : OBGIOPClientWorker(client, transport), destroy_(false), exception_(0)
{
    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;
    }
}

OBGIOPClientWorkerThreaded::~OBGIOPClientWorkerThreaded()
{
    assert(destroy_);
    delete exception_;
}

// ----------------------------------------------------------------------
// OBGIOPClientWorkerThreaded public member implementations
// ----------------------------------------------------------------------

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

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

    //
    // 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_TRUE);
    
    //
    // 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();
}

CORBA_ULong
OBGIOPClientWorkerThreaded::sendReceiveTimeout(CORBA_ULong reqId,
					       OBBuffer& buf, bool& swap,
					       GIOP_ReplyStatusType& status,
					       CORBA_Long t)
{
    JTCSynchronized synchronized(*this);

    //
    // Check whether there is an exception
    //
    if(exception_)
    {
	//
	// Restoring the buffer is not necessary here, since it has
	// not been converted into an OCI buffer yet
	//

	//
	// Destroy the worker and raise the exception
	//
	client_ -> destroyWorker(this);
	exception_ -> _raise();
    }

    //
    // Convert simple buffer to OCI buffer
    // Add the OCI buffer
    //
    OCI_Buffer_var b = new OCI_Buffer;
    b -> buf().consume(buf);
    add(b);

    //
    // Return the reply
    //
    // Upon communication failure, the buffer may not be restored, in
    // order to guarantee "at most once" semantics
    //
    return receiveTimeout(reqId, buf, swap, status, t);
}

CORBA_ULong
OBGIOPClientWorkerThreaded::receiveTimeout(CORBA_ULong reqId, OBBuffer& buf,
					   bool& swap,
					   GIOP_ReplyStatusType& status,
					   CORBA_Long t)
{
    //
    // Yield if zero timeout
    //
    if(t == 0)
	JTCThread::yield();

    JTCSynchronized synchronized(*this);
    {
	CORBA_ULong offset = 0;

	if(t < 0)
	{
	    //
	    // Wait for reply or exception, without timeout
	    //
	    while(!exception_ &&
		  (offset = findReply(reqId, buf, swap, status)) == 0)
	    {
		try
		{
		    wait();
		}
		catch(const JTCInterruptedException&)
		{
		}
	    }
	}
	else if(t == 0)
	{
	    //
	    // Zero timeout, don't wait for reply
	    //
	    if(!exception_)
		offset = findReply(reqId, buf, swap, status);

	    if(offset == 0)
		return 0; // Timeout
	}
	else
	{
	    //
	    // Get current time
	    //
#ifdef WIN32
	    struct _timeb tb1;
	    _ftime(&tb1);
#else
	    timeval tv1;
	    gettimeofday(&tv1, 0);
#endif
	    CORBA_Long millis = 0;

	    //
	    // Wait for reply or exception, with timeout
	    //
	    while(!exception_ &&
		  (offset = findReply(reqId, buf, swap, status)) == 0)
	    {
		try
		{
		    wait(t);
		}
		catch(const JTCInterruptedException&)
		{
		}

		//
		// Calculate time needed
		//
#ifdef WIN32
		struct _timeb tb2;
		_ftime(&tb2);
		millis = (tb2.time - tb1.time) * 1000;
		millis += tb2.millitm;
		millis -= tb1.millitm;
#else
		timeval tv2;
		gettimeofday(&tv2, 0);
		millis = (tv2.tv_sec - tv1.tv_sec) * 1000;
		millis += tv2.tv_usec / 1000;
		millis -= tv1.tv_usec / 1000;
#endif
		//
		// Check if time limit has been exceeded
		//
		if(millis >= t)
		    return 0; // Timeout
	    }
	}

	//
	// Check whether there is an exception
	//
	if(exception_)
	{
	    //
	    // Destroy the worker and raise the exception
	    //
	    client_ -> destroyWorker(this);
	    exception_ -> _raise();
	}

	assert(offset);
	return offset;
    }
}

void
OBGIOPClientWorkerThreaded::sendNonBlocking(OBBuffer& buf)
{
    JTCSynchronized synchronized(*this);

    //
    // Check whether there is an exception
    //
    if(exception_)
    {
	//
	// Restoring the buffer is not necessary here, since it has
	// not been converted into an OCI buffer yet
	//

	//
	// Destroy the worker and raise the exception
	//
	client_ -> destroyWorker(this);
	exception_ -> _raise();
    }

    //
    // Convert simple buffer to OCI buffer
    // Add the OCI buffer
    //
    OCI_Buffer_var b = new OCI_Buffer;
    b -> buf().consume(buf);
    add(b);

    //
    // Nothing else to do here
    //
}

void
OBGIOPClientWorkerThreaded::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
	    //
	    transport_ -> send(buf, true);
	    assert(buf -> is_full());
	}
	catch(const CORBA_SystemException& ex)
	{
	    {
		JTCSynchronized synchronized(*this);

		//
		// It is possible that the sender thread has already set
		// an exception
		//
		if(!exception_)
		{
		    exception_ =
			CORBA_SystemException::_narrow(ex._OB_clone());
		    assert(exception_);
		}
		
		notifyAll();
		
		//
		// The worker may *not* be removed from the client,
		// otherwise the client will not detect the communication
		// failure.
		//
		//client_ -> destroyWorker(this);
		destroy();
	    }

	    return;
	}
    }

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

void
OBGIOPClientWorkerThreaded::receiverRun()
{
    while(!destroy_)
    {
	try
	{
	    //
	    // Setup the incoming message buffer
	    //
	    assert(CORBA_is_nil(buf_));
	    setupBuffer();
	    
	    //
	    // Receive header, blocking
	    //
	    transport_ -> receive(buf_, true);
	    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
		//
		transport_ -> receive(buf_, true);
		assert(buf_ -> is_full());
	    }
	    
	    //
	    // Execute the message
	    //
	    {
		JTCSynchronized synchronized(*this);

		if(!destroy_)
		{
		    execute();
		    notifyAll();
		}
	    }
	}
	catch(const CORBA_SystemException& ex)
	{
	    {
		JTCSynchronized synchronized(*this);

		//
		// It is possible that the sender thread has already set
		// an exception
		//
		if(!exception_)
		{
		    exception_ =
			CORBA_SystemException::_narrow(ex._OB_clone());
		    assert(exception_);
		}
		
		notifyAll();

		//
		// The worker may *not* be removed from the client,
		// otherwise the client will not detect the communication
		// failure.
		//
		//client_ -> destroyWorker(this);
		destroy();
	    }
	    	    
	    return;
	}
    }
}

// ----------------------------------------------------------------------
// OBGIOPClientWorkerThreaded::SenderThread public member implementations
// ----------------------------------------------------------------------

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

// ----------------------------------------------------------------------
// OBGIOPClientWorkerThreaded::ReceiverThread public member implementations
// ----------------------------------------------------------------------

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

#endif
