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

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

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

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

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

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

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

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

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

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

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

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

#ifndef HAVE_NO_EXPLICIT_TEMPLATES
template class OBObjVar< OBGIOPServerStarter >;
template class OBObjForSeq< OBGIOPServerStarter >;
template class OBObjVar< OBGIOPServerStarterBlocking >;
template class OBObjForSeq< OBGIOPServerStarterBlocking >;
template class OBObjVar< OBGIOPServerStarterReactive >;
template class OBObjForSeq< OBGIOPServerStarterReactive >;
#ifdef HAVE_JTC
template class OBObjVar< OBGIOPServerStarterThreaded >;
template class OBObjForSeq< OBGIOPServerStarterThreaded >;
#endif
#else
#ifdef HAVE_PRAGMA_DEFINE
#pragma define(OBObjVar< OBGIOPServerStarter >)
#pragma define(OBObjForSeq< OBGIOPServerStarter >)
#pragma define(OBObjVar< OBGIOPServerStarterBlocking >)
#pragma define(OBObjForSeq< OBGIOPServerStarterBlocking >)
#pragma define(OBObjVar< OBGIOPServerStarterReactive >)
#pragma define(OBObjForSeq< OBGIOPServerStarterReactive >)
#ifdef HAVE_JTC
#pragma define(OBObjVar< OBGIOPServerStarterThreaded >)
#pragma define(OBObjForSeq< OBGIOPServerStarterThreaded >)
#endif
#endif
#endif

// ----------------------------------------------------------------------
// OBGIOPServerStarter private and protected member implementation
// ----------------------------------------------------------------------

void
OBGIOPServerStarter::destroyAllWorkers()
{
    //
    // Destroy all workers. Make sure that any further destructions do
    // not block.
    //
    OBGIOPServerWorkerSeq workers;

    {
#ifdef HAVE_JTC
	JTCSynchronized synchronized(workersMutex_);
#endif
        workers = workers_;
        workers_.length(0);
    }

    for(CORBA_ULong i = 0 ; i < workers.length() ; i++)
	workers[i] -> destroy();
}

// ----------------------------------------------------------------------
// OBGIOPServerStarter constructor and destructor
// ----------------------------------------------------------------------

OBGIOPServerStarter::OBGIOPServerStarter(OCI_Acceptor_ptr acceptor,
					 bool enabled)
    : acceptor_(OCI_Acceptor::_duplicate(acceptor))
{
    //cout << "OBGIOPServerStarter" << endl;
    //
    // Only start the acceptor listening if enabled
    //
    try
    {
        if(enabled)
            acceptor_ -> listen();
    }
    catch(const CORBA_SystemException&)
    {
        _OB_setRef(0);
        throw;
    }
}

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

    //
    // If workers_ is not empty, destroy() was not called
    //
    assert(workers_.length() == 0);
}

// ----------------------------------------------------------------------
// OBGIOPServerStarter public member implementation
// ----------------------------------------------------------------------

void
OBGIOPServerStarter::destroyWorker(OBGIOPServerWorker_ptr worker)
{
    assert(!CORBA_is_nil(worker));

    //
    // Destroy the worker. Make sure that any further destructions do
    // not block.
    //
    OBGIOPServerWorker_var w;
    
    {
#ifdef HAVE_JTC
	JTCSynchronized synchronized(workersMutex_);
#endif
	    
	for(CORBA_ULong i = 0 ; i < workers_.length() ; i++)
	    if(workers_[i].in() == worker)
	    {
		w = workers_[i]._retn();
		workers_.remove(i);
		break;
	    }
    }

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

bool
OBGIOPServerStarter::startWorker()
{
    return false;
}

// ----------------------------------------------------------------------
// OBGIOPServerStarterBlocking constructor and destructor
// ----------------------------------------------------------------------

OBGIOPServerStarterBlocking::OBGIOPServerStarterBlocking(
    OCI_Acceptor_ptr acceptor, bool enabled)
    : OBGIOPServerStarter(acceptor, enabled)
{
}

OBGIOPServerStarterBlocking::~OBGIOPServerStarterBlocking()
{
}

// ----------------------------------------------------------------------
// OBGIOPServerStarterBlocking public member implementation
// ----------------------------------------------------------------------

void
OBGIOPServerStarterBlocking::destroy()
{
    //
    // Close the acceptor
    //
    acceptor_ -> close();

    //
    // Destroy all workers
    //
    destroyAllWorkers();
}

bool
OBGIOPServerStarterBlocking::startWorker()
{
    try
    {
	OCI_Transport_var transport = acceptor_ -> accept();
	assert(!CORBA_is_nil(transport));

	OBGIOPServerWorkerBlocking_var worker =
	    new OBGIOPServerWorkerBlocking(this, transport);
	workers_.append(OBGIOPServerWorker::_duplicate(worker));
	
	while(true)
	{
	    if(workers_.length() == 0)
		break;

	    assert(workers_.length() == 1);
	    worker -> getRequest();

	    if(workers_.length() == 0)
		break;

	    assert(workers_.length() == 1);
	    worker -> sendReply();
	}
    }
    catch(const CORBA_NO_PERMISSION&)
    {
	//
	// Ignore NO_PERMISSION exceptions
	//
    }

    return true;
}

// ----------------------------------------------------------------------
// OBGIOPServerStarterReactive constructor and destructor
// ----------------------------------------------------------------------

OBGIOPServerStarterReactive::OBGIOPServerStarterReactive(
    OCI_Acceptor_ptr acceptor, bool enabled)
    : OBGIOPServerStarter(acceptor, enabled), destroy_(false)
{
    //
    // Register as read event handler
    //
    OCI_Handle handle = acceptor_ -> handle();
    assert(handle >= 0);
    OBReactor::instance() -> registerHandler(this, OBEventRead, handle);
}

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

// ----------------------------------------------------------------------
// OBGIOPServerStarterReactive public member implementation
// ----------------------------------------------------------------------

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

    //
    // Don't destroy twice
    //
    if(destroy_)
	return;
    
    //
    // Set the destroy flag
    //
    destroy_ = true;
    
    //
    // Unregister as event handler
    //
    OBReactor::instance() -> unregisterHandler(this);

    //
    // Close the acceptor
    //
    acceptor_ -> close();

    //
    // Destroy all workers
    //
    destroyAllWorkers();
}

void
OBGIOPServerStarterReactive::handleEvent(OBMask mask)
{
    //
    // Handle read events
    //
    assert(mask == OBEventRead);

    try
    {
	//
	// Get new transport
	//
	OCI_Transport_var transport = acceptor_ -> accept();
	assert(!CORBA_is_nil(transport));

	//
	// Create new worker
	//
	{
#ifdef HAVE_JTC
	    JTCSynchronized synchronized(workersMutex_);
#endif
	    workers_.append(new OBGIOPServerWorkerReactive(this, transport));
	}
    }
    catch(const CORBA_NO_PERMISSION&)
    {
	//
	// Ignore NO_PERMISSION exceptions
	//
    }
}

#ifdef HAVE_JTC

// ----------------------------------------------------------------------
// OBGIOPServerStarterThreaded constructor and destructor
// ----------------------------------------------------------------------

OBGIOPServerStarterThreaded::OBGIOPServerStarterThreaded(
    OCI_Acceptor_ptr acceptor, bool enabled)
    : OBGIOPServerStarter(acceptor, enabled), destroy_(false)
{
    try
    {
	OB_DEC_TRY_FIX

	//
	// Create starter thread
	//
	starterThread_ = new StarterThread(this);

	//
	// Start the starter thread if not disabled
	//
	if(enabled)
	    starterThread_ -> start();
    }
    catch(...)
    {
	_OB_setRef(0);
	throw;
    }
}

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

// ----------------------------------------------------------------------
// OBGIOPServerStarterThreaded public member implementation
// ----------------------------------------------------------------------

void
OBGIOPServerStarterThreaded::destroy()
{
    //
    // Don't destroy twice
    //
    if(destroy_)
	return;
    
    //
    // Set the destroy flag
    //
    destroy_ = true;

    //
    // Unblock threads blocking in accept()
    //
    acceptor_ -> unblock_threads();

    //
    // Join with the starter thread
    //
    assert(starterThread_ -> getId() != JTCThread::currentThread() -> getId());
    while(starterThread_ -> isAlive())
    {
	try
	{
	    starterThread_ -> join();
	}
	catch(const JTCInterruptedException&)
	{
	}
    }

    //
    // Close the acceptor
    //
    acceptor_ -> close();

    //
    // Destroy all workers
    //
    destroyAllWorkers();
}

void
OBGIOPServerStarterThreaded::starterRun()
{
    while(!destroy_)
    {
	try
	{
	    //
	    // Get new transport
	    //
	    OCI_Transport_var transport = acceptor_ -> accept();
	    assert(!CORBA_is_nil(transport));
	    
	    //
	    // Check if thread should be stopped and destroyed
	    //
	    if(destroy_)
	    {
		transport -> close(CORBA_TRUE);
		break;
	    }
	    
	    //
	    // Create new worker
	    //
	    if(!CORBA_is_nil(transport))
	    {
		JTCSynchronized synchronized(workersMutex_);

		try
		{
		    OB_DEC_TRY_FIX
		    workers_.append(new
			OBGIOPServerWorkerThreaded(this, transport));
		}
		catch(const JTCException&)
		{
		    //
		    // Thread creation failed
		    //
		    transport -> close(CORBA_TRUE);
		    return;
		}
	    }
	}
	catch(const CORBA_NO_PERMISSION&)
	{
	    //
	    // Ignore NO_PERMISSION exceptions
	    //
	}
	catch(const CORBA_SystemException&)
	{
	    //
	    // Destroy the worker
	    //
            // Not sure what to do here. Really destroy? Ignore?
	    //destroy();
	}
    }
}

// ----------------------------------------------------------------------
// OBGIOPServerStarterThreaded::SenderThread public member implementations
// ----------------------------------------------------------------------

void
OBGIOPServerStarterThreaded::StarterThread::run()
{
    starter_ -> starterRun();
    // Break cyclic object dependency
    starter_ = OBGIOPServerStarterThreaded::_nil();
}

#endif
