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

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

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

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

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

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

#ifndef HAVE_NO_EXPLICIT_TEMPLATES
template class OBObjVar< OBGIOPClient >;
template class OBObjForSeq< OBGIOPClient >;
#else
#ifdef HAVE_PRAGMA_DEFINE
#pragma define(OBObjVar< OBGIOPClient >)
#pragma define(OBObjForSeq< OBGIOPClient >)
#endif
#endif

// ----------------------------------------------------------------------
// OBGIOPClient private and protected member implementations
// ----------------------------------------------------------------------

void
OBGIOPClient::addHeaders(CORBA_Object_ptr p, const char* op, CORBA_ULong reqId,
			 bool ow, OBBuffer& buf)
{
    CORBA_Octet* oct = buf.data;
    GIOP_MessageHeader hdr =
    {
	{ 'G', 'I', 'O', 'P' }, { 1, 0 },
	OBEndian, GIOP_Request, buf.len - 12
    };
    OBMarshal(hdr, oct);

    const OCI_ObjectKey& key = p -> _OB_key();
    if(key.length() == 0) // If key is not cached, extract from IOR
    {
	OCI_ObjectKey_var k = connector_ -> is_usable(p -> _OB_ior());
	p -> _OB_key(k);
	assert(key.length());
    }

    GIOP_RequestHeader reqH;
    reqH.request_id = reqId;
    reqH.response_expected = !ow;
#ifdef OB_CLEAR_MEM
    reqH.reserved[0] = 0;
    reqH.reserved[1] = 0;
    reqH.reserved[2] = 0;
#endif
    reqH.object_key = key;
    reqH.operation = op;
    OBMarshal(reqH, oct);
}

OBGIOPClientWorker_ptr
OBGIOPClient::getWorker(CORBA_Long t)
{
#ifdef HAVE_JTC
    JTCSynchronized synchronized(workerMutex_);
#endif

    if(CORBA_is_nil(worker_))
    {
	//
	// Create new transport, using the connector
	//
	// TODO: In reactive mode, this is not correct. A non-blocking
	// connect() operation should be added to the OCI. Then a
	// GIOPClientStarterReactive class should be written, that
	// registers with the Reactor in order to listen to write
	// events on the filedescriptor of the non-blocking Connector
	// (handle() must be added to the Connector). If it receives a
	// write event, it gets the Transport from the Connector by
	// calling an operation complete_connect() (must also be
	// added). Then it creates an GIOPClientWorkerReactive with
	// the newly created Transport.
	//
	// For symetry reasons, GIOPClientStarterBlocking and
	// GIOPClientStarterThreaded should also be added, even though
	// these classes only have a trivial functionality. Or perhaps
	// the GIOPClientStarterThreaded tries to connect() in the
	// backgound? Just an idea...
	//
	OCI_Transport_var transport;
	if(t >= 0)
	{
	    transport = connector_ -> connect_timeout((CORBA_ULong)t);

	    //
	    // Was there a timeout?
	    //
	    if(CORBA_is_nil(transport))
		throw CORBA_NO_RESPONSE();
	}
	else
	{
	    transport = connector_ -> connect();
	}
	
	//
	// Create new worker
	//
	switch(CORBA_ORB::conc_model())
	{
	case CORBA_ORB::ConcModelBlocking:
	    worker_ = new OBGIOPClientWorkerBlocking(this, transport);
	    break;
	    
	case CORBA_ORB::ConcModelReactive:
	    worker_ = new OBGIOPClientWorkerReactive(this, transport);
	    break;
	    
#ifdef HAVE_JTC
	case CORBA_ORB::ConcModelThreaded:
	    worker_ = new OBGIOPClientWorkerThreaded(this, transport);
	    break;
#endif
	}
    }

    return OBGIOPClientWorker::_duplicate(worker_);
}

// ----------------------------------------------------------------------
// OBGIOPClient constructor and destructor
// ----------------------------------------------------------------------

OBGIOPClient::OBGIOPClient(OCI_Connector_ptr connector)
    : connector_(OCI_Connector::_duplicate(connector))
{
    //cout << "OBGIOPClient" << endl;
}

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

    //
    // If worker_ is not nil, destroy() was not called
    //
    assert(CORBA_is_nil(worker_));
}

// ----------------------------------------------------------------------
// OBGIOPClient public member implementation
// ----------------------------------------------------------------------

void
OBGIOPClient::destroy()
{
    //
    // Destroy the worker. Make sure that any further destructions do
    // not block.
    //
    OBGIOPClientWorker_var w;

    {
#ifdef HAVE_JTC
	JTCSynchronized synchronized(workerMutex_);
#endif
	w = worker_._retn();
    }

    if(!CORBA_is_nil(w))
	w -> destroy();
}

void
OBGIOPClient::destroyWorker(OBGIOPClientWorker_ptr worker)
{
    assert(!CORBA_is_nil(worker));
	
    //
    // Destroy the worker. Make sure that any further destructions do
    // not block.
    //
    OBGIOPClientWorker_var w;

    {
#ifdef HAVE_JTC
	JTCSynchronized synchronized(workerMutex_);
#endif
	if(worker_.in() == worker)
	    w = worker_._retn();
    }

    if(!CORBA_is_nil(w))
	w -> destroy();
}

OCI_Connector_ptr
OBGIOPClient::connector()
{
    return OCI_Connector::_duplicate(connector_);
}

OCI_Transport_ptr
OBGIOPClient::transport()
{
    OBGIOPClientWorker_var worker = getWorker(-1);
    return worker -> transport();
}

CORBA_ULong
OBGIOPClient::offset(CORBA_Object_ptr p, const char* op)
{
    CORBA_ULong offset = 12;
    
    const OCI_ObjectKey& key = p -> _OB_key();
    if(key.length() == 0) // If key is not cached, extract from IOR
    {
	OCI_ObjectKey_var k = connector_ -> is_usable(p -> _OB_ior());
	p -> _OB_key(k);
	assert(key.length());
    }

    GIOP_RequestHeader reqH;
#ifdef OB_CLEAR_MEM
    reqH.request_id = 0;
    reqH.response_expected = true;
    reqH.reserved[0] = 0;
    reqH.reserved[1] = 0;
    reqH.reserved[2] = 0;
#endif
    reqH.object_key = key;
    reqH.operation = op;
    
    OBMarshalCount(reqH, offset);
    
    return offset;
}

CORBA_ULong
OBGIOPClient::request(CORBA_Object_ptr p, const char* op, OBBuffer& buf,
		      bool& swap, bool& exc, bool& fwd,
		      CORBA_Long t, bool retry)
{
    GIOP_ReplyStatusType status;
    CORBA_ULong offset;

    while(true)
    {
	//
	// Get the worker
	//
	OBGIOPClientWorker_var worker = getWorker(t);

	//
	// Add headers
	//
	CORBA_ULong reqId = worker -> getRequestId();
	addHeaders(p, op, reqId, false, buf);
	
	try
	{
	    //
	    // Send request and receive reply, with timeout
	    //
	    offset = worker -> sendReceiveTimeout(reqId, buf, swap, status, t);
	}
	catch(const CORBA_COMM_FAILURE&)
	{
	    //
	    // Retry once on communication failure if the retry
	    // parameter is set to true
	    //
	    if(retry && buf.len)
	    {
		retry = false;
		continue;
	    }

	    throw;
	}
	catch(const CORBA_TRANSIENT&)
	{
	    //
	    // Retry on transient failure
	    //
	    if(buf.len)
		continue;

	    throw;
	}
	
	//
	// Was there a timeout?
	//
	if(offset == 0)
	{
	    assert(t >= 0);
	    
	    //
	    // Mark the request ID as discarded and throw a timeout
	    // exception
	    //
	    worker -> discardRequest(reqId);
	    throw CORBA_NO_RESPONSE();
	}

	break;
    }

    //
    // Evaluate reply
    //
    switch(status)
    {
    case GIOP_NO_EXCEPTION:
	exc = false;
	fwd = false;
	break;
	
    case GIOP_USER_EXCEPTION:
	exc = true;
	fwd = false;
	break;
	
    case GIOP_SYSTEM_EXCEPTION:
    {
	const CORBA_Octet* coct = buf.data + offset;
	OBUnmarshalSystemExceptionThrow(coct, swap);
	break;
    }
	
    case GIOP_LOCATION_FORWARD:
	exc = false;
	fwd = true;
	break;
    }

    return offset;
}

void
OBGIOPClient::oneway(CORBA_Object_ptr p, const char* op, OBBuffer& buf,
		     bool retry)
{
    while(true)
    {
	//
	// Get the worker
	//
	// TODO: The -1 timeout argument is wrong! What really needs
	// to be done is to use a 0 timeout here. But then the
	// connect() in getWorker() will fail in (nearly) all cases,
	// meaning that all oneways to servers to which the client has
	// not yet connected will be discarded! The right solution is
	// to decouple the buffers from the worker, so that the oneway
	// can be put into the send buffers, even if no worker exits.
	//
	// The decoupling of worker and buffers has to be done anyway,
	// in order to more easily support bi-directional GIOP. That
	// is, it should be possible that both, a GIOPClientWorker and
	// a GIOPServerWorker have access to the same buffers.
	//
	OBGIOPClientWorker_var worker = getWorker(-1);
	
	//
	// Add headers
	//
	CORBA_ULong reqId = worker -> getRequestId();
	addHeaders(p, op, reqId, true, buf);
	
	try
	{
	    //
	    // Send request non-blocking
	    //
	    worker -> sendNonBlocking(buf);
	}
	catch(const CORBA_COMM_FAILURE&)
	{
	    //
	    // Retry once on communication failure if the retry
	    // parameter is set to true
	    //
	    if(retry && buf.len)
	    {
		retry = false;
		continue;
	    }
	    
	    throw;
	}
	catch(const CORBA_TRANSIENT&)
	{
	    //
	    // Retry on transient failure
	    //
	    if(buf.len)
		continue;

	    throw;
	}
	
	break;
    }
}

void
OBGIOPClient::deferred(CORBA_Object_ptr p, const char* op, OBBuffer& buf,
		       CORBA_ULong& reqId, bool retry)
{
    while(true)
    {
	//
	// Get the worker
	//
	// TODO: The -1 argument is wrong here. See the comments in
	// "oneway" above.
	//
	OBGIOPClientWorker_var worker = getWorker(-1);
	
	//
	// Add headers
	//
	reqId = worker -> getRequestId();
	addHeaders(p, op, reqId, false, buf);
	
	try
	{
	    //
	    // Send request non-blocking
	    //
	    // This should really be sendTimeout, but I don't have such an
	    // operation right now.
	    //
	    worker -> sendNonBlocking(buf);
	}
	catch(const CORBA_COMM_FAILURE&)
	{
	    //
	    // Retry once on communication failure if the retry
	    // parameter is set to true
	    //
	    if(retry && buf.len)
	    {
		retry = false;
		continue;
	    }
	    
	    throw;
	}
	catch(const CORBA_TRANSIENT&)
	{
	    //
	    // Retry on transient failure
	    //
	    if(buf.len)
		continue;

	    throw;
	}

	break;
    }
}

CORBA_ULong
OBGIOPClient::response(CORBA_ULong reqId, OBBuffer& buf,
		       bool& swap, bool& exc, bool& fwd, CORBA_Long t)
{
    GIOP_ReplyStatusType status;
    CORBA_ULong offset;

    while(true)
    {
        //
	// Get the worker
	//
	OBGIOPClientWorker_var worker = getWorker(t);
	
	try
	{
	    //
	    // Receive reply, with timout
	    //
	    offset = worker -> receiveTimeout(reqId, buf, swap, status, t);
	}
	catch(const CORBA_TRANSIENT&)
	{
	    //
	    // Retry on transient failure
	    //
	    continue;
	}

	//
	// Was there a timeout?
	//
	if(offset == 0)
	{
	    assert(t >= 0);
	    
	    //
	    // Mark the request ID as discarded and throw a timeout
	    // exception
	    //
	    worker -> discardRequest(reqId);
	    throw CORBA_NO_RESPONSE();
	}

	break;
    }

    //
    // Evaluate reply
    //
    switch(status)
    {
    case GIOP_NO_EXCEPTION:
	exc = false;
	fwd = false;
	break;
	
    case GIOP_USER_EXCEPTION:
	exc = true;
	fwd = false;
	break;
	
    case GIOP_SYSTEM_EXCEPTION:
    {
	const CORBA_Octet* coct = buf.data + offset;
	OBUnmarshalSystemExceptionThrow(coct, swap);
	break;
    }
	
    case GIOP_LOCATION_FORWARD:
	exc = false;
	fwd = true;
	break;
    }

    return offset;
}

CORBA_ULong
OBGIOPClient::poll(CORBA_ULong reqId, OBBuffer& buf,
		   bool& swap, bool& exc, bool& fwd)
{
    GIOP_ReplyStatusType status;
    CORBA_ULong offset;

    while(true)
    {
	//
	// Get the worker
	//
	// TODO: The -1 argument is wrong here. See the comments in
	// "oneway" above.
	//
	OBGIOPClientWorker_var worker = getWorker(-1);

	try
	{
	    //
	    // Receive reply, with timeout
	    //
	    offset = worker -> receiveTimeout(reqId, buf, swap, status, 0);
	}
	catch(const CORBA_TRANSIENT&)
	{
	    //
	    // Retry on transient failure
	    //
	    continue;
	}
	
	//
	// Was there a timeout?
	//
	if(offset == 0)
	    return 0;
	
	break;
    }

    //
    // Evaluate reply
    //
    switch(status)
    {
    case GIOP_NO_EXCEPTION:
	exc = false;
	fwd = false;
	break;
	
    case GIOP_USER_EXCEPTION:
	exc = true;
	fwd = false;
	break;
	
    case GIOP_SYSTEM_EXCEPTION:
    {
	const CORBA_Octet* coct = buf.data + offset;
	OBUnmarshalSystemExceptionThrow(coct, swap);
	break;
    }
	
    case GIOP_LOCATION_FORWARD:
	exc = false;
	fwd = true;
	break;
    }

    return offset;
}
