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

#ifndef OB_GIOP_CLIENT_WORKER_H
#define OB_GIOP_CLIENT_WORKER_H

//
// The OBGIOPClientWorker class
//
class OBGIOPClientWorker;
typedef OBGIOPClientWorker* OBGIOPClientWorker_ptr;
void OBDuplicate(OBGIOPClientWorker_ptr);
void OBRelease(OBGIOPClientWorker_ptr);
typedef OBObjVar< OBGIOPClientWorker > OBGIOPClientWorker_var;

class OBGIOPClientWorker : public OBRefCount OB_COMMA_MONITOR
{
protected:

    static CORBA_ULong nextRequestId_; // Next request ID
    static OBFixSeq< CORBA_ULong > discReqIdSeq_; // Discarded request IDs
#ifdef HAVE_JTC
    static JTCMutex requestIdMutex_; // Monitor for request ID handling
#endif

    OBGIOPClient_var client_; // The client
    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

public: // For some compilers the following struct must be public
    struct Reply
    {
	CORBA_ULong reqId;
	OBBuffer buf;
	bool swap;
	GIOP_ReplyStatusType status;
	CORBA_ULong offset;
	Reply* next;
    };
protected:
    Reply* head_; // I can't use OBVarSeq<> because buf must not
                  // be copied (operator=()) for performance reasons

    //
    // Add a "message error" message
    //
    void addMessageError();

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

    //
    // Find a reply
    // Check for a reply
    //
    CORBA_ULong findReply(CORBA_ULong, OBBuffer&, bool&,
			  GIOP_ReplyStatusType&);
    CORBA_Boolean checkReply(CORBA_ULong);

public:

    OBGIOPClientWorker(OBGIOPClient_ptr, OCI_Transport_ptr);
    virtual ~OBGIOPClientWorker();

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

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

    //
    // Get the OCI transport
    //
    OCI_Transport_ptr transport();

    //
    // Get a request id
    // Discard a request
    //
    CORBA_ULong getRequestId()
    {
#ifdef HAVE_JTC
	JTCSynchronized synchronized(requestIdMutex_);
#endif
	return nextRequestId_++;
    }
    void discardRequest(CORBA_ULong reqId)
    {
#ifdef HAVE_JTC
	JTCSynchronized synchronized(requestIdMutex_);
#endif
	discReqIdSeq_.append(reqId);
    }

    //
    // Send a request and receive a reply, with timeout
    // Receive a reply, with timeout
    // Send a request, non-blocking
    //
    virtual CORBA_ULong sendReceiveTimeout(CORBA_ULong, OBBuffer&, bool&,
					   GIOP_ReplyStatusType&,
					   CORBA_Long) = 0;
    virtual CORBA_ULong receiveTimeout(CORBA_ULong, OBBuffer&, bool&,
				       GIOP_ReplyStatusType&, CORBA_Long) = 0;
    virtual void sendNonBlocking(OBBuffer&) = 0;
};

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

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

//
// The OBGIOPClientWorkerBlocking class
//
class OBGIOPClientWorkerBlocking;
typedef OBGIOPClientWorkerBlocking* OBGIOPClientWorkerBlocking_ptr;
void OBDuplicate(OBGIOPClientWorkerBlocking_ptr);
void OBRelease(OBGIOPClientWorkerBlocking_ptr);
typedef OBObjVar< OBGIOPClientWorkerBlocking > OBGIOPClientWorkerBlocking_var;

class OBGIOPClientWorkerBlocking : public OBGIOPClientWorker
{
public:

    OBGIOPClientWorkerBlocking(OBGIOPClient_ptr, OCI_Transport_ptr);
    virtual ~OBGIOPClientWorkerBlocking();

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

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

    //
    // Send a request and receive a reply, with timeout
    // Receive a reply, with timeout
    // Send a request, non-blocking
    //
    virtual CORBA_ULong sendReceiveTimeout(CORBA_ULong, OBBuffer&, bool&,
					   GIOP_ReplyStatusType&, CORBA_Long);
    virtual CORBA_ULong receiveTimeout(CORBA_ULong, OBBuffer&, bool&,
				       GIOP_ReplyStatusType&, CORBA_Long);
    virtual void sendNonBlocking(OBBuffer&);
};

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

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

//
// The OBGIOPClientWorkerReactive class
//
class OBGIOPClientWorkerReactive;
typedef OBGIOPClientWorkerReactive* OBGIOPClientWorkerReactive_ptr;
void OBDuplicate(OBGIOPClientWorkerReactive_ptr);
void OBRelease(OBGIOPClientWorkerReactive_ptr);
typedef OBObjVar< OBGIOPClientWorkerReactive > OBGIOPClientWorkerReactive_var;

class OBGIOPClientWorkerReactive : public OBGIOPClientWorker,
				   public OBEventHandler
{
protected:

    bool destroy_; // True if destroy() was called
    CORBA_SystemException* exception_; // The buffered exception

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

public:

    OBGIOPClientWorkerReactive(OBGIOPClient_ptr, OCI_Transport_ptr);
    ~OBGIOPClientWorkerReactive();

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

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

    //
    // Send a request and receive a reply, with timeout
    // Receive a reply, with timeout
    // Send a request, non-blocking
    //
    virtual CORBA_ULong sendReceiveTimeout(CORBA_ULong, OBBuffer&, bool&,
					   GIOP_ReplyStatusType&, CORBA_Long);
    virtual CORBA_ULong receiveTimeout(CORBA_ULong, OBBuffer&, bool&,
				       GIOP_ReplyStatusType&, CORBA_Long);
    virtual void sendNonBlocking(OBBuffer&);

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

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

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

#ifdef HAVE_JTC

//
// The OBGIOPClientWorkerThreaded class
//
class OBGIOPClientWorkerThreaded;
typedef OBGIOPClientWorkerThreaded* OBGIOPClientWorkerThreaded_ptr;
void OBDuplicate(OBGIOPClientWorkerThreaded_ptr);
void OBRelease(OBGIOPClientWorkerThreaded_ptr);
typedef OBObjVar< OBGIOPClientWorkerThreaded > OBGIOPClientWorkerThreaded_var;

class OBGIOPClientWorkerThreaded : public OBGIOPClientWorker
{
protected:

    bool destroy_; // True if destroy() was called
    CORBA_SystemException* exception_; // The buffered exception

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

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

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

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

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

public:

    OBGIOPClientWorkerThreaded(OBGIOPClient_ptr, OCI_Transport_ptr);
    ~OBGIOPClientWorkerThreaded();

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

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

    //
    // Send a request and receive a reply, with timeout
    // Receive a reply, with timeout
    // Send a request, non-blocking
    //
    virtual CORBA_ULong sendReceiveTimeout(CORBA_ULong, OBBuffer&, bool&,
					   GIOP_ReplyStatusType&, CORBA_Long);
    virtual CORBA_ULong receiveTimeout(CORBA_ULong, OBBuffer&, bool&,
				       GIOP_ReplyStatusType&, CORBA_Long);
    virtual void sendNonBlocking(OBBuffer&);

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

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

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

#endif

#endif
