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

#ifndef OB_GIOP_SERVER_WORKER_H
#define OB_GIOP_SERVER_WORKER_H

//
// The OBGIOPServerWorker class
//
class OBGIOPServerWorker;
typedef OBGIOPServerWorker* OBGIOPServerWorker_ptr;
void OBDuplicate(OBGIOPServerWorker_ptr);
void OBRelease(OBGIOPServerWorker_ptr);
typedef OBObjVar< OBGIOPServerWorker > OBGIOPServerWorker_var;
typedef OBObjSeq< OBGIOPServerWorker > OBGIOPServerWorkerSeq;

class OBGIOPServerWorker : public OBRefCount
{
    //
    // Hide copy-constructor and asignment operator
    //
    OBGIOPServerWorker(const OBGIOPServerWorker&);
    void operator=(const OBGIOPServerWorker&);

    void displayWarning(const char*, const char*, OCI_TransportInfo_ptr);

protected:

    OBGIOPServerStarter_var starter_; // The server starter
    OCI_Transport_var transport_; // The transport
    OCI_BufferSeq outSeq_; // The outgoing message buffers
    OCI_Buffer_var buf_; // The incoming message buffer
    GIOP_MessageHeader hdr_; // The incoming message header
    
    //
    // Add a "message error" message
    // Add a "close connection" message
    //
    void addMessageError();
    void addCloseConnection();

    //
    // Setup buf_ so that a new incoming message can be received
    //
    void setupBuffer();

    //
    // Extract the hdr_ from buf_
    //
    void extractHeader();

    //
    // Add outgoing message
    //
    virtual void add(OCI_Buffer_ptr);

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

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

    //
    // Execute incoming message
    //
    void execute();

    //
    // Dispatch a request
    //
    virtual void dispatchRequest(OBBuffer&, bool);

public:

    OBGIOPServerWorker(OBGIOPServerStarter_ptr, OCI_Transport_ptr);
    virtual ~OBGIOPServerWorker();

    static inline OBGIOPServerWorker_ptr _duplicate(OBGIOPServerWorker_ptr p)
    { if(p) p -> _OB_incRef(); return p; }
    static inline OBGIOPServerWorker_ptr _nil()
    { return 0; }

    //
    // Destroy the worker
    //
    virtual void destroy() = 0;
};

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

inline CORBA_Boolean
CORBA_is_nil(OBGIOPServerWorker_ptr p)
{
    return p == 0;
}

//
// The OBGIOPServerWorkerBlocking class
//
class OBGIOPServerWorkerBlocking;
typedef OBGIOPServerWorkerBlocking* OBGIOPServerWorkerBlocking_ptr;
void OBDuplicate(OBGIOPServerWorkerBlocking_ptr);
void OBRelease(OBGIOPServerWorkerBlocking_ptr);
typedef OBObjVar< OBGIOPServerWorkerBlocking > OBGIOPServerWorkerBlocking_var;

class OBGIOPServerWorkerBlocking : public OBGIOPServerWorker
{
    //
    // Hide copy-constructor and asignment operator
    //
    OBGIOPServerWorkerBlocking(const OBGIOPServerWorkerBlocking&);
    void operator=(const OBGIOPServerWorkerBlocking&);

public:

    OBGIOPServerWorkerBlocking(OBGIOPServerStarterBlocking_ptr,
			       OCI_Transport_ptr);
    virtual ~OBGIOPServerWorkerBlocking();

    static inline OBGIOPServerWorkerBlocking_ptr
    _duplicate(OBGIOPServerWorkerBlocking_ptr p)
    { if(p) p -> _OB_incRef(); return p; }
    static inline OBGIOPServerWorkerBlocking_ptr _nil()
    { return 0; }

    //
    // Destroy the worker
    //
    virtual void destroy();

    //
    // Get a request
    // Send a reply
    //
    void getRequest();
    void sendReply();
};

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

inline CORBA_Boolean
CORBA_is_nil(OBGIOPServerWorkerBlocking_ptr p)
{
    return p == 0;
}

//
// The OBGIOPServerWorkerReactive class
//
class OBGIOPServerWorkerReactive;
typedef OBGIOPServerWorkerReactive* OBGIOPServerWorkerReactive_ptr;
void OBDuplicate(OBGIOPServerWorkerReactive_ptr);
void OBRelease(OBGIOPServerWorkerReactive_ptr);
typedef OBObjVar< OBGIOPServerWorkerReactive > OBGIOPServerWorkerReactive_var;

class OBGIOPServerWorkerReactive : public OBGIOPServerWorker,
				   public OBEventHandler
{
    //
    // Hide copy-constructor and asignment operator
    //
    OBGIOPServerWorkerReactive(const OBGIOPServerWorkerReactive&);
    void operator=(const OBGIOPServerWorkerReactive&);

protected:

    bool destroy_; // True if destroy() was called

    //
    // Add outgoing message
    //
    virtual void add(OCI_Buffer_ptr);

public:

    OBGIOPServerWorkerReactive(OBGIOPServerStarterReactive_ptr,
			       OCI_Transport_ptr);
    virtual ~OBGIOPServerWorkerReactive();

    static inline OBGIOPServerWorkerReactive_ptr
    _duplicate(OBGIOPServerWorkerReactive_ptr p)
    { if(p) p -> _OB_incRef(); return p; }
    static inline OBGIOPServerWorkerReactive_ptr _nil()
    { return 0; }

    //
    // Destroy the worker
    //
    virtual void destroy();

    //
    // Handle event
    //
    virtual void handleEvent(OBMask);
};

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

inline CORBA_Boolean
CORBA_is_nil(OBGIOPServerWorkerReactive_ptr p)
{
    return p == 0;
}

#ifdef HAVE_JTC

//
// The OBGIOPServerWorkerThreaded class
//
class OBGIOPServerWorkerThreaded;
typedef OBGIOPServerWorkerThreaded* OBGIOPServerWorkerThreaded_ptr;
void OBDuplicate(OBGIOPServerWorkerThreaded_ptr);
void OBRelease(OBGIOPServerWorkerThreaded_ptr);
typedef OBObjVar< OBGIOPServerWorkerThreaded > OBGIOPServerWorkerThreaded_var;

class OBGIOPServerWorkerThreaded : public OBGIOPServerWorker, public JTCMonitor
{
protected:

    bool destroy_; // True if destroy() was called

    //
    // The sender thread
    //
    class SenderThread : public JTCThread
    {
	OBGIOPServerWorkerThreaded_var worker_;

    public:
	
	SenderThread(OBGIOPServerWorkerThreaded_ptr worker)
	    : worker_(OBGIOPServerWorkerThreaded::_duplicate(worker)) { }
	virtual ~SenderThread() { }
	virtual void run();
    };
    friend class SenderThread;
    JTCThreadHandle senderThread_;

    //
    // The receiver thread
    //
    class ReceiverThread : public JTCThread
    {
	OBGIOPServerWorkerThreaded_var worker_;

    public:
	
	ReceiverThread(OBGIOPServerWorkerThreaded_ptr worker)
	    : worker_(OBGIOPServerWorkerThreaded::_duplicate(worker)) { }
	virtual ~ReceiverThread() { }
	virtual void run();
    };
    friend class ReceiverThread;
    JTCThreadHandle receiverThread_;

    //
    // Monitor for "single threaded" thread model, which allows
    // only one thread for request execution.
    //
    static JTCMutex singleThreadedMutex_;

    //
    // Helper class for dispatching a request in case of a "thread per
    // request" concurrency model
    //
    class Dispatcher : public JTCThread
    {
	OBGIOPServerWorkerThreaded_var worker_;
	OBBuffer buf_;
	bool swap_;
        OCI_Transport_var transport_;

    public:

	Dispatcher(OBGIOPServerWorkerThreaded_ptr w, OBBuffer& b, bool s)
	    : worker_(OBGIOPServerWorkerThreaded::_duplicate(w)), swap_(s)
	{
	    buf_.consume(b);
	}

	virtual ~Dispatcher()
	{
	}
       
	virtual void run()
	{
	    worker_ -> OBGIOPServerWorker::dispatchRequest(buf_, swap_);
	}
    };
    friend class Dispatcher;

    //
    // Helper class for dispatching a request in case of a "thread
    // pool" concurrency model
    //
public: // For some compilers the following class must be public
    class ThreadPool : public JTCMonitor
    {
	static ThreadPool* instance_; // The thread pool singleton
	static JTCMutex instanceMutex_; // The thread pool singleton mutex
	bool destroy_; // True if destroy() was called
	CORBA_ULong num_; // The number of threads in the pool
	JTCThreadHandle* handles_; // Handles for the threads in the pool

public: // For some compilers the following struct must be public
	struct Request
	{
	    OBGIOPServerWorkerThreaded_var worker;
	    OBBuffer buf;
	    bool swap;
	    Request* next;
	};
private:
	Request* head_; // I can't use OBVarSeq<> because buf must not
	                // be copied (operator=()) for performance reasons

	void get(Request&); // Pulls a request from head_

public: // Must be public for some compilers, otherwise the "friend
        // class ThreadPool::Dispatcher;" declaration produces warnings
        class Dispatcher : public JTCThread
	{
	    ThreadPool* threadPool_;
	    
	public:
	    
	    Dispatcher(ThreadPool* threadPool)
		: threadPool_(threadPool)
	    {
	    }
	    
	    virtual void
	    run()
	    {
		Request req;
		    
		do
		{
		    threadPool_ -> get(req);

		    if(!CORBA_is_nil(req.worker))
		    {
			req.worker ->
			    OBGIOPServerWorker::dispatchRequest(req.buf,
								req.swap);
		    }
		}
		while(!CORBA_is_nil(req.worker));
	    }
	};
private:
	friend class Dispatcher;

	ThreadPool();

    public:

	virtual ~ThreadPool();

	static ThreadPool* instance();
	static void destroyInstance();

	void destroy();
	void add(OBGIOPServerWorkerThreaded_ptr, OBBuffer&, bool);
    };
    friend class ThreadPool::Dispatcher;
private:

    //
    // Add outgoing message
    // Remove outgoing message
    //
    virtual void add(OCI_Buffer_ptr);
    virtual OCI_Buffer_ptr remove();

    //
    // Dispatch a request
    //
    virtual void dispatchRequest(OBBuffer&, bool);

public:

    OBGIOPServerWorkerThreaded(OBGIOPServerStarterThreaded_ptr,
			       OCI_Transport_ptr);
    ~OBGIOPServerWorkerThreaded();

    static inline OBGIOPServerWorkerThreaded_ptr
    _duplicate(OBGIOPServerWorkerThreaded_ptr p)
    { if(p) p -> _OB_incRef(); return p; }
    static inline OBGIOPServerWorkerThreaded_ptr _nil()
    { return 0; }

    //
    // Destroy the worker
    //
    virtual void destroy();

    //
    // Run method for receiver and sender thread
    //
    void senderRun();
    void receiverRun();
};

inline void
CORBA_release(OBGIOPServerWorkerThreaded_ptr p)
{
    if(p)
	p -> _OB_decRef();
}

inline CORBA_Boolean
CORBA_is_nil(OBGIOPServerWorkerThreaded_ptr p)
{
    return p == 0;
}

#endif

#endif
