// **********************************************************************
//
// 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/Principal.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.h>
#include <OB/Reactor.h>
#include <OB/Timer.h>
#include <OB/Hashtable.h>
#include <OB/HashtableI.h>
#include <OB/Hashers.h>
#include <OB/OCI_IIOP.h>
#include <OB/Properties.h>

#include <OCI_impl.h> // For new OCI_AccRegistry_impl

#include <GIOPServer.h>

#include <stdio.h>

//
// The BOA - only one instance is allowed
//
CORBA_BOA_ptr CORBA_BOA::boa_ = CORBA_BOA::_nil();

//
// The concurrency model
//
CORBA_BOA::ConcModel CORBA_BOA::concModel_ = CORBA_BOA::ConcModelReactive;
CORBA_ULong CORBA_BOA::concModelThreadPool_ = 10;

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

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

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

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

#ifndef HAVE_NO_EXPLICIT_TEMPLATES
template class OBObjVar< CORBA_BOA >;
template class OBObjForSeq< CORBA_BOA >;
// template class OBObjSeq< CORBA_BOA >;
// template class OBSeqVar< OBObjSeq< CORBA_BOA > >;

template class OBHashtable< OBFixSeq< CORBA_Octet >, CORBA_Object_var,
                            OBLockStrategyNull>;
template class OBHashtable< CORBA_Object_ptr,
                            OBVarSeq< OBFixSeq< CORBA_Octet > >,
                            OBLockStrategyNull>;
#else
#ifdef HAVE_PRAGMA_DEFINE
#pragma define(OBObjVar< CORBA_BOA >)
#pragma define(OBObjForSeq< CORBA_BOA >)
// #pragma define(OBObjSeq< CORBA_BOA >)
// #pragma define(OBSeqVar< OBObjSeq< CORBA_BOA > >)

#pragma define(OBHashtable< OBFixSeq< CORBA_Octet >, CORBA_Object_var,
	                   OBLockStrategyNull >)
#pragma define(OBHashtable< CORBA_Object_ptr, \
                            OBVarSeq< OBFixSeq< CORBA_Octet > >,
	                    OBLockStrategyNull >)

#endif
#endif

// ----------------------------------------------------------------------
// For CORBA_BOA::KeyHash
// ----------------------------------------------------------------------

CORBA_ULong
OBHash(const CORBA_Object_ptr obj)
{
    return (CORBA_ULong)(unsigned long)obj;
}

bool
OBHashCompare(const CORBA_Object_ptr o1, const CORBA_Object_ptr o2)
{
    return o1 == o2;
}

// ----------------------------------------------------------------------
// BOA constructor and destructor
// ----------------------------------------------------------------------

CORBA_BOA::CORBA_BOA()
    : num_(0)
{
    //
    // Check if there is already a BOA singleton
    //
    assert(boa_ == 0);

    //
    // Set the BOA singleton to this instance
    //
    boa_ = this;

    //
    // Create the registry
    //
    registry_ = new OCI_AccRegistry_impl;

    //
    // Create hash table for the connected servants
    //
    impls_ = new CORBA_BOA::ObjectHash(1023);
    keys_ = new CORBA_BOA::ServantKeyHash(1023);
}

CORBA_BOA::~CORBA_BOA()
{
    CORBA_ULong i;

    //
    // Destroy all servers
    //
    for(i = 0 ; i < servers_.length() ; i++)
	servers_[i] -> destroy();

    //
    // TODO: This is a hack!
    //
#ifdef HAVE_JTC
    void OBDestroyThreadPool();
    OBDestroyThreadPool();
#endif

    //
    // Destroy the hash table
    //
    delete impls_;
    delete keys_;

    //
    // Set BOA singleton to nil
    //
    boa_ = CORBA_BOA::_nil();
}

// ----------------------------------------------------------------------
// BOA public member implementation
// ----------------------------------------------------------------------

CORBA_Object_ptr
CORBA_BOA::create(const CORBA_ReferenceData& /*id*/,
		  CORBA_InterfaceDef_ptr /*intf*/,
		  CORBA_ImplementationDef_ptr /*impl*/)
{
    assert_nca(false, OBNCANotImpl);
    return CORBA_Object::_nil();
}

void
CORBA_BOA::dispose(CORBA_Object_ptr /*obj*/)
{
    assert_nca(false, OBNCANotImpl);
}

CORBA_ReferenceData* 
CORBA_BOA::get_id(CORBA_Object_ptr /*obj*/)
{
    assert_nca(false, OBNCANotImpl);
    return 0;
}

void
CORBA_BOA::change_implementation(CORBA_Object_ptr /*obj*/,
				 CORBA_ImplementationDef_ptr /*impl*/)
{
    assert_nca(false, OBNCANotImpl);
}

CORBA_Principal_ptr
CORBA_BOA::get_principal(CORBA_Object_ptr /*obj*/,
			 CORBA_Environment_ptr /*ev*/)
{
    assert_nca(false, OBNCANotImpl);
    return CORBA_Principal::_nil();
}

void
CORBA_BOA::impl_is_ready(CORBA_ImplementationDef_ptr /*implDef*/)
{
#ifdef HAVE_JTC
    JTCSynchronized synchronized(serversMonitor_);
#endif

    //
    // Initialize servers if not already done
    //
    if(init_servers())
        return; // All done

    //
    // If reactive mode is selected (client- or server-side), then we
    // must dispatch events using the reactor. Otherwise we can wait
    // until all servers are released.
    //
    if(CORBA_ORB::conc_model() == CORBA_ORB::ConcModelReactive ||
       CORBA_BOA::conc_model() == CORBA_BOA::ConcModelReactive)
    {
	OBReactor::instance() -> dispatch();
    }
#ifdef HAVE_JTC
    else
    {
	while(servers_.length() > 0)
	{
	    try
	    {
		serversMonitor_.wait();
	    }
	    catch(const JTCInterruptedException&)
	    {
	    }
	}
    }
#endif
}

void
CORBA_BOA::deactivate_impl(CORBA_ImplementationDef_ptr)
{
#ifdef HAVE_JTC
    JTCSynchronized synchronized(serversMonitor_);
#endif

    //
    // Destroy servers
    //
    destroy_servers();
    assert(servers_.length() == 0);

    //
    // If reactive mode is selected (client- or server-side), then we
    // must tell the reactor to stop dispatching events. Otherwise we
    // notify the thread that waits for the servers to be released.
    //
    if(CORBA_ORB::conc_model() == CORBA_ORB::ConcModelReactive ||
       CORBA_BOA::conc_model() == CORBA_BOA::ConcModelReactive)
    {
	OBReactor::instance() -> stop();
    }
#ifdef HAVE_JTC
    else
    {
	serversMonitor_.notify();
    }
#endif
}

void
CORBA_BOA::obj_is_ready(CORBA_Object_ptr obj, CORBA_ImplementationDef_ptr)
{
    connect(obj);
}

void
CORBA_BOA::deactivate_obj(CORBA_Object_ptr obj)
{
    disconnect(obj);
}

OCI_AccRegistry_ptr
CORBA_BOA::get_acc_registry()
{
    return OCI_AccRegistry::_duplicate(registry_);
}

CORBA_BOA::ConcModel
CORBA_BOA::conc_model()
{
    return concModel_;
}

void
CORBA_BOA::conc_model(CORBA_BOA::ConcModel model)
{
    concModel_ = model;
}

CORBA_ULong
CORBA_BOA::conc_model_thread_pool()
{
    return concModelThreadPool_;
}

void
CORBA_BOA::conc_model_thread_pool(CORBA_ULong num)
{
    concModelThreadPool_ = num;
}

void
CORBA_BOA::create_new_servers()
{
#ifdef HAVE_JTC
    JTCSynchronized synchronized(serversMonitor_);
#endif

    assert(!CORBA_is_nil(registry_));

    //
    // This creates servers for any new acceptors. It is known that
    // acceptors are only appended to the acceptors list, so we can
    // check the current length of the servers list, and the new
    // length of the acceptors list, and create servers for any newly
    // appended acceptors.
    //
    // Another way of coding this would be to look at the tags of the
    // known acceptors, and known servers can create new server for an
    // unknown acceptor tag. Of course, this assumes that the set of
    // tags is unique (must be true anyway).
    //
    OCI_AcceptorSeq_var acceptors = registry_ -> get_acceptors();
    if(servers_.length() < acceptors -> length())
    {
        int save = servers_.length();
        servers_.length(acceptors -> length());
        for(CORBA_ULong i = save ; i < servers_.length() ; i++)
        {
            servers_[i] = new OBGIOPServer(acceptors[i]);
        }
    }
}

bool
CORBA_BOA::init_servers()
{
#ifdef HAVE_JTC
    JTCSynchronized synchronized(serversMonitor_);
#endif

    create_new_servers();

    bool allDone = true;
    for(CORBA_ULong i = 0 ; i < servers_.length() ; i++)
    {
	bool done = servers_[i] -> run();
	allDone = allDone && done;
    }
    
    return allDone;
}

void
CORBA_BOA::destroy_servers()
{
#ifdef HAVE_JTC
    JTCSynchronized synchronized(serversMonitor_);
#endif

    //
    // Destroy and release all servers
    //
    for(CORBA_ULong i = 0 ; i < servers_.length() ; i++)
	servers_[i] -> destroy();

    servers_.length(0);
}

void
CORBA_BOA::connect(CORBA_Object_ptr obj)
{
    if(obj -> _is_local())
    {
#ifdef HAVE_JTC
	JTCSynchronized synchronized(implsMutex_);
#endif
	    
	//
	// Check if implementation is already connected
	//
        if(keys_ -> contains(obj))
            return;

        //
        // Get current time
        //
        timeval tv = OBTimerList::timeNow();
        
	//
	// Makeup object key. Object key is 0 + tv_sec + tv_usec +
	// running count. We need the tv_usec to account for a server
	// restarting in the same second
	//
	OCI_ObjectKey key;
	key.length(16);
	CORBA_Octet* oct = key.data();
	OBMarshal((CORBA_ULong)0, oct);
        OBMarshal((CORBA_ULong)tv.tv_sec, oct);
        OBMarshal((CORBA_ULong)tv.tv_usec, oct);
	OBMarshal(num_++, oct);
	
	//
	// Add key to the IOR
	//
	IOP_IOR ior = obj -> _OB_ior();
	ior.type_id = obj -> _OB_typeId();
	registry_ -> add_profiles(key, ior);
	obj -> _OB_ior(ior);
	
	//
	// Add to implementation sequence
	//
        impls_ -> put(key, CORBA_Object::_duplicate(obj));

        //
        // Add to the object -> key(s) map
        //
        CORBA_BOA::ServantKeySeq seq;
        seq.append(key);
        keys_ -> put(obj, seq);
    }
}

void
CORBA_BOA::connect(CORBA_Object_ptr obj, const char* name)
{
    if(obj -> _is_local())
    {
        //
        // Check for empty string. This is not allowed.
        //
	assert_nca(name, OBNCANullString);
        if(*name == '\0')
            throw CORBA_BAD_PARAM();

#ifdef HAVE_JTC
	JTCSynchronized synchronized(implsMutex_);
#endif

	//
	// Check if implementation is already connected
	//
        if(keys_ -> contains(obj))
            return;
	
	//
	// Makeup object key
	//
	// I don't use string marshal functions here because I don't
	// want to have a string length encoded. That would cause byte
	// order dependencies.
	//
	int len = strlen(name) + 1;
	OCI_ObjectKey key(len);
	key.length(len);
	strcpy((char*)key.data(), name);

	//
	// Check if key already exists
	//
        if(impls_ -> contains(key))
            throw CORBA_INV_IDENT();
	
	//
	// Add key to the IOR
	//
	IOP_IOR ior = obj -> _OB_ior();
	ior.type_id = obj -> _OB_typeId();
	registry_ -> add_profiles(key, ior);
	obj -> _OB_ior(ior);
	
	//
	// Add to implementation sequence
	//
        impls_ -> put(key, CORBA_Object::_duplicate(obj));

        //
        // Record the key
        //
        CORBA_BOA::ServantKeySeq seq(3);
        seq.append(key);

	//
	// Add a second key for compatibility with OB 2.0 named
	// objects
	//
	{
	    //
	    // The following code comes from OB2.0
	    //
	    //
	    // Makeup object key
	    //
	    // I don't use string marshal functions here because I don't
	    // want to have a string length encoded. That would cause byte
	    // order dependencies.
	    //
	    // Note: If this is changed, ORB::get_inet_object() must also
	    // be changed
	    //
	    const char* ob = "OB/NAME";
	    int obLen = strlen(ob) + 1;
	    int nameLen = strlen(name) + 1;

	    OCI_ObjectKey key(obLen + nameLen);
	    key.length(obLen + nameLen);
	    strcpy((char*)key.data(), ob);
	    strcpy((char*)key.data() + obLen, name);

	    //
	    // Add to implementation sequence
	    //
            impls_ -> put(key, CORBA_Object::_duplicate(obj));

            //
            // Record the key
            //
            seq.append(key);
	}

	//
	// Add a third key for compatibility with the INS object
	// URLs
	//
	{
	    CORBA_ULong len = strlen(name);
	    OCI_ObjectKey key(len);
	    key.length(len);
	    memcpy((char*)key.data(), name, len);

	    //
	    // Add to implementation sequence
	    //
            impls_ -> put(key, CORBA_Object::_duplicate(obj));

            //
            // Record the key
            //
            seq.append(key);
	}

	//
	// Add to the object -> key(s) map
	//
	keys_ -> put(obj, seq);
    }
}

void
CORBA_BOA::disconnect(CORBA_Object_ptr obj)
{
    if(obj -> _is_local())
    {
#ifdef HAVE_JTC
	JTCSynchronized synchronized(implsMutex_);
#endif

	obj -> _OB_ior(IOP_IOR()); // Remove IOR

        CORBA_BOA::ServantKeySeq seq;
        if(keys_ -> get(obj, seq))
        {
            keys_ -> remove(obj);
            for(CORBA_ULong i = 0; i < seq.length(); ++i)
                impls_ -> remove(seq[i]);
        }
    }
}

void
CORBA_BOA::disable_acceptor(CORBA_ULong tag)
{
    //
    // Initialize servers if not already done
    //
    create_new_servers();
    for(CORBA_ULong i = 0; i < servers_.length(); ++i)
    {
        if(servers_[i] -> acceptor() -> tag() == tag)
        {
            servers_[i] -> disable();
        }
    }
}

//
// Display statistics on the internal implementation hash table
//
void
CORBA_BOA::display_stats(ostream& os)
{
#ifdef HAVE_JTC
    JTCSynchronized synchronized(implsMutex_);
#endif
    impls_ -> displayStats(os);
}

CORBA_BOA::ObjectSeq
CORBA_BOA::get_connected_servants()
{
    ObjectSeq servants;

#ifdef HAVE_JTC
    JTCSynchronized synchronized(implsMutex_);
#endif
    CORBA_BOA::ObjectHash::Enumerator keys = impls_ -> keys();
    for(CORBA_ULong i = 0; i < keys.length(); i++)
    {
        CORBA_Object_var obj;
        impls_ -> get(keys[i], obj);
        assert(!CORBA_is_nil(obj));
        CORBA_ULong j;
        for(j = 0; j < servants.length(); j++)
            if(servants[j] -> _is_equivalent(obj))
                break;
        if(j >= servants.length())
            servants.append(CORBA_Object::_duplicate(obj));
    }
    return servants;
}

//
// Set the IIOP port from the port number defined in the IIOP profile
// in an Object
//
/*static*/ void
CORBA_BOA::set_iiop_port_from_object(CORBA_Object_ptr obj)
{
    //
    // If the object is nil, do nothing.
    //
    if(CORBA_is_nil(obj))
	return;

    //
    // Get the OCI connector info.
    //
    OCI_ConnectorInfo_var info = obj -> _get_oci_connector_info();

    //
    // Narrow to the IIOP connector info.
    //
    OCI_IIOP_ConnectorInfo_var iiopInfo =
	OCI_IIOP_ConnectorInfo::_narrow(info);

    //
    // If this is an IIOP connector info then rip out the remote port
    // number, and set the ooc.boa.port property to this port.
    //
    if(!CORBA_is_nil(iiopInfo))
    {
	//
	// Allocate 10 bytes so that we don't need to reallocate when
	// we add the port.
	//
	CORBA_String_var s = CORBA_string_alloc(10);
	s[0] = '\0';
	s += iiopInfo -> remote_port();

	//
	// Set the property.
	//
	OBProperties::instance() -> setProperty("ooc.boa.port", s);
    }
}

CORBA_BOA_ptr
CORBA_BOA::_OB_instance()
{
    return CORBA_BOA::_duplicate(boa_);
}

CORBA_BOA_ptr
CORBA_BOA::_OB_instance_nodup()
{
    return boa_;
}

CORBA_Object_ptr
CORBA_BOA::_OB_find(const OCI_ObjectKey& key)
{
#ifdef HAVE_JTC
    JTCSynchronized synchronized(implsMutex_);
#endif

    CORBA_Object_var obj;
    impls_ -> get(key, obj);
    return obj._retn();
}

CORBA_Object_ptr
CORBA_BOA::_OB_find(const IOP_IOR& ior)
{
#ifdef HAVE_JTC
    JTCSynchronized synchronized(implsMutex_);
#endif

    OCI_ObjectKey_var key = registry_ -> is_local(ior);

    CORBA_Object_var obj;
    if(key -> length() > 0)
	impls_ -> get(key, obj);
    return obj._retn();
}

OBDispatchStatus
CORBA_BOA::_OB_dispatch(const OCI_ObjectKey& key, const char* op,
			OBBuffer& buf, bool sw,
			CORBA_ULong offIn, CORBA_ULong offOut)
{
    CORBA_Object_var impl = _OB_find(key);
    
    if(!CORBA_is_nil(impl))
    {
	return impl -> _OB_dispatch(op, buf, sw, offIn, offOut);
    }
    
    //
    // Accept _non_existent and _not_existent. (See the ORB Core
    // revision task force mailing list for a discussion why.)
    //
    if(strcmp(op, "_non_existent") == 0 || strcmp(op, "_not_existent") == 0)
    {
        CORBA_Boolean r = true;
	
	CORBA_ULong cnt = offOut;
	OBMarshalCount(r, cnt);
	
	buf.alloc(cnt);
	CORBA_Octet* o = buf.data + offOut;
	OBMarshal(r, o);
	
	return OBDispatchStatusOK;
    }
    else
	return OBDispatchStatusNoObject; // Object does not exist
}

// ----------------------------------------------------------------------
// BOA Initialization
// ----------------------------------------------------------------------

/* See IIOPInit.cpp */
