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

#include <OB/CORBA.h>
#include <OB/Util.h>
#include <OB/Properties.h>

#include <OB/CosEventChannelAdmin_skel.h>

#include <ProxyConsumer_impl.h>
#include <ProxyState.h>
#include <EC_ProxyPullConsumer_impl.h>
#include <EC_SupplierAdmin_impl.h>
#include <EC_EventChannel_impl.h>
#include <PropagatorBaseThread.h>
#include <Event_impl.h>
#include <QoS.h>
#include <EventUtil.h>

#include <stdlib.h> // for atoi, ...

// ----------------------------------------------------------------------
// AnyPullThread implementation
// ----------------------------------------------------------------------

class AnyPullerThread : public PropagatorBaseThread
{
    CosEventChannelAdmin_ProxyPullConsumer_var consumer_; // The proxy.

    CosEventComm_PullSupplier_var supplier_; // The supplier.

    EC_EventChannel_impl* channel_; // The event channel.

    CORBA_ULong pullInterval_; // Interval between successive pulls.

    bool hasDelay_; // Is there a delay between each pull call?

public:

    AnyPullerThread(CosEventComm_PullSupplier_ptr supplier,
		    CosEventChannelAdmin_ProxyPullConsumer_ptr consumer,
		    CORBA_ULong id,
		    EC_EventChannel_impl* channel,
		    CORBA_ULong pullInterval,
		    CORBA_ULong maxRetries,
		    CORBA_ULong retryTimeout,
		    CORBA_Double retryMultiplier)
        : PropagatorBaseThread(id, maxRetries, retryTimeout, retryMultiplier),
          consumer_(CosEventChannelAdmin_ProxyPullConsumer::
                    _duplicate(consumer)),
          supplier_(CosEventComm_PullSupplier::_duplicate(supplier)),
          channel_(channel),
          pullInterval_(pullInterval),
          hasDelay_(pullInterval > 0)
        
    {
    }

    ~AnyPullerThread()
    {
#ifdef OB_TRACE
        OBMessageViewer::instance() -> trace(2, "Destroy AnyPullerThread");
#endif
    }

    virtual void run()
    {
        while(!done_)
        {
            try
            {
                CORBA_Any_var any = supplier_ -> pull();
                resetRetry();
                Event_impl* event = new Event_impl(*any, id_);
                channel_ -> receive(event);
                if(hasDelay_)
                    delay(pullInterval_);
            }
            catch(const CORBA_COMM_FAILURE&)
            {
                if(!retry())
                {
                    done_ = true;
                    try
                    {
                        consumer_ -> disconnect_pull_consumer();
                    }
                    catch(...)
                    {
                    }
                }
            }
            catch(...)
            {
                done_ = true;
                try
                {
                    consumer_ -> disconnect_pull_consumer();
                }
                catch(...)
                {
                }
            }
        }

        //
        // Break the circular relationship
        //
        consumer_ = CosEventChannelAdmin_ProxyPullConsumer::_nil();

        //
        // We are also not interested in this either.
        //
        supplier_ = CosEventComm_PullSupplier::_nil();
    }
};

// ----------------------------------------------------------------------
// EC_ProxyPullConsumer_impl constructor/destructor
// ----------------------------------------------------------------------

EC_ProxyPullConsumer_impl::EC_ProxyPullConsumer_impl(
    CORBA_ORB_ptr orb,
    EC_EventChannel_impl* channel,
    EC_SupplierAdmin_impl* admin,
    CORBA_ULong id)
    : orb_(CORBA_ORB::_duplicate(orb)),
      channel_(channel),
      admin_(admin),
      state_(Disconnected),
      id_(id)
{
#ifdef OB_TRACE
    if(OBMessageViewer::instance() -> getTraceLevel() >= 2)
    {
        CORBA_String_var desc = EventUtil::getConnectionAddrDescription(orb_);
        CORBA_String_var msg = CORBA_string_dup("Create ProxyPullConsumer(");
        msg += id_;
        msg += ") from ";
        msg += desc;
        OBMessageViewer::instance() -> trace(2, msg);
    }
#endif
}

EC_ProxyPullConsumer_impl::~EC_ProxyPullConsumer_impl()
{
#ifdef OB_TRACE
    if(OBMessageViewer::instance() -> getTraceLevel() >= 2)
    {
        CORBA_String_var msg = CORBA_string_dup("Destroy ProxyPullConsumer(");
        msg += id_;
        msg += ')';
        OBMessageViewer::instance() -> trace(2, msg);
    }
#endif

}

// ----------------------------------------------------------------------
// EC_ProxyPullConsumer_impl public member implementation
// ----------------------------------------------------------------------

void
EC_ProxyPullConsumer_impl::connect_pull_supplier(
    CosEventComm_PullSupplier_ptr supplier)
{
    JTCSynchronized sync(*this);

    if(state_ == Destroyed)
	throw CORBA_OBJECT_NOT_EXIST();
    
    if(CORBA_is_nil(supplier))
	throw CORBA_BAD_PARAM();

    if(state_ == Connected)
	throw CosEventChannelAdmin_AlreadyConnected();

    supplier_ = CosEventComm_PullSupplier::_duplicate(supplier);

    state_ = Connected;

    assert(!puller_);    

    //
    // Resolve the QoS params
    //
    OBProperties* properties = OBProperties::instance();
    CORBA_ULong pullInterval =
        atoi(properties -> getProperty(EventQoS::PULL_INTERVAL));
    CORBA_ULong retryTimeout =
        atoi(properties -> getProperty(EventQoS::RETRY_TIMEOUT));
    CORBA_Double retryMultiplier =
        atof(properties -> getProperty(EventQoS::RETRY_MULTIPLIER));
    CORBA_ULong maxRetries =
        atoi(properties -> getProperty(EventQoS::MAX_RETRIES));
    
    puller_ = new AnyPullerThread(supplier, this, id_, channel_,
                                  pullInterval, maxRetries, retryTimeout,
                                  retryMultiplier);
    puller_ -> start();
}

void
EC_ProxyPullConsumer_impl::disconnect_pull_consumer()
{
    //
    // We first must check the state. This is to account for the case
    // that disconnect_pull_consumer is called concurrently.
    //
    ProxyState saveState;
    {
	JTCSynchronized sync(*this);
	if(state_ == Destroyed)
	    return;
	saveState = state_;
	state_ = Destroyed;
    }

    admin_ -> removeProxy(this);

    //
    // Stop the puller, if running.
    //
    if(puller_)
        puller_ -> halt();
    
    if(saveState == Connected)
    {
	try
	{
	    supplier_ -> disconnect_pull_supplier();
	}
	catch(const CORBA_SystemException&)
	{
	    // A non-existent supplier doesn't matter in this context.
	}
    }

    //
    // Disconnect the servant from the ORB.
    //
    orb_ -> disconnect(this);
}

//
// Internal methods
//
void
EC_ProxyPullConsumer_impl::disconnect()
{
    {
	JTCSynchronized sync(*this);
	
	if(state_ == Destroyed)
	    return;
	state_ = Destroyed;
    }
    
    //
    // Stop the puller, wait for it to complete, if running.
    //
    if(puller_)
    {
        puller_ -> halt();
        while(puller_ -> isAlive())
        {
            try
            {
                puller_ -> join();
            }
            catch(const JTCInterruptedException&)
            {
            }
        }
    }

    //
    // Disconnect the servant from the ORB.
    //
    orb_ -> disconnect(this);
}

void
EC_ProxyPullConsumer_impl::release()
{
    CORBA_release(this);
}
