// **********************************************************************
//
// 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/CosTypedEventChannelAdmin_skel.h>

#include <ProxySupplier_impl.h>
#include <ProxyState.h>
#include <TEC_ProxyPushSupplier_impl.h>
#include <TEC_TypedConsumerAdmin_impl.h>
#include <PropagatorBaseThread.h>
#include <PusherBaseThread.h>
#include <QoS.h>
#include <EventUtil.h>
#include <Event_impl.h>

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

// ----------------------------------------------------------------------
// TypedPusherThread implementation
// ----------------------------------------------------------------------

class TypedPusherThread : public PusherBaseThread
{
    PushResult pushEvent(Event_impl* event)
    {
        assert(event != 0);
        assert(!CORBA_is_nil(consumer_));

        //
        // Create a DII request.
        //
        CORBA_Request_var request = consumer_ -> _request(event -> op());
        
        //
        // Insert each of the arguments.
        //
        for(CORBA_ULong i = 0; i < event -> args() -> count(); ++i)
        {
            CORBA_NamedValue_ptr nv = event -> args() -> item(i);
            CORBA_Any& arg = request -> add_in_arg(nv -> name());
            arg = *nv -> value();
        }

        PushResult rc;

        try
        {
            //
            // Invoke the request.
            //
            request -> invoke();

            //
            // Did we receive an exception?
            //
            CORBA_Exception* e = request -> env() -> exception();
            if(e != 0)
                e -> _raise();

            rc = Success;
        }
        catch(const CORBA_COMM_FAILURE&)
        {
            //
            // If the exception is COMM_FAILURE then we retry, failure
            // otherwise.
            //
            rc = Retry;
        }
        catch(const CORBA_Exception&)
        {
            rc = Failed;
        }

        return rc;
    }

public:

    TypedPusherThread(CosEventComm_PushConsumer_ptr consumer,
                      CosEventChannelAdmin_ProxyPushSupplier_ptr proxy,
                      CORBA_ULong id,
                      CORBA_ULong maxEvents,
                      CORBA_ULong maxRetries,
                      CORBA_ULong retryTimeout,
                      CORBA_Double retryMultiplier)
        : PusherBaseThread(consumer, proxy, id, maxEvents, maxRetries,
                           retryTimeout, retryMultiplier)
    {
    }
};

// ----------------------------------------------------------------------
// TEC_ProxyPushSupplier_impl constructor/destructor
// ----------------------------------------------------------------------

TEC_ProxyPushSupplier_impl::TEC_ProxyPushSupplier_impl(
    CORBA_ORB_ptr orb,
    TEC_TypedConsumerAdmin_impl* admin,
    CORBA_ULong id)
    
    : orb_(CORBA_ORB::_duplicate(orb)),
      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 TypedProxyPushSupplier(");
        msg += id_;
        msg += ") from ";
        msg += desc;
        OBMessageViewer::instance() -> trace(2, msg);
    }
#endif
}

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

// ----------------------------------------------------------------------
// TEC_ProxyPushSupplier_impl public member implementation
// ----------------------------------------------------------------------

void
TEC_ProxyPushSupplier_impl::connect_push_consumer(
    CosEventComm_PushConsumer_ptr consumer)
{
    JTCSynchronized sync(*this);

    if(state_ == Destroyed)
        throw CORBA_OBJECT_NOT_EXIST();
    
    if(CORBA_is_nil(consumer))
	throw CORBA_BAD_PARAM();
    
    if(state_ == Connected)
	throw CosEventChannelAdmin_AlreadyConnected();

    consumer_ = CosEventComm_PushConsumer::_duplicate(consumer);

    state_ = Connected;

    //
    // Resolve the QoS params, create the TypedPusherThread, and start
    // the thread running.
    //
    assert(!pusher_);

    OBProperties* properties = OBProperties::instance();
    CORBA_ULong maxEvents =
        atoi(properties -> getProperty(EventQoS::MAX_EVENTS));
    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));
    
    pusher_ = new TypedPusherThread(consumer, this, id_, maxEvents,
                                    maxRetries, retryTimeout, retryMultiplier);
    pusher_ -> start();
}

void
TEC_ProxyPushSupplier_impl::disconnect_push_supplier()
{
    ProxyState saveState;

    //
    // We first must check the state. This is to account for the case
    // that disconnect_push_supplier is called concurrently. We don't
    // simply lock the entire method because that can cause a deadlock
    // between receive and this method.
    //
    {
        JTCSynchronized sync(*this);
        if(state_ == Destroyed)
            return;
        saveState = state_;
        state_ = Destroyed;
    }

    admin_ -> removeProxy(this);

    //
    // Ask the pusher to stop, if running.
    //
    if(pusher_)
        pusher_ -> halt();

    //
    // If we're connected then tell the consumer.
    //
    if(saveState == Connected)
    {
        try
        {
            consumer_ -> disconnect_push_consumer();
        }                                              
        catch(const CORBA_SystemException&)
        {
        }
    }

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

void
TEC_ProxyPushSupplier_impl::receive(Event_impl* event)
{
    JTCSynchronized sync(*this);

    //
    // Don't accumulate events if we're not yet connected.
    //
    if(state_ != Connected)
        return;

    pusher_ -> newEvent(event);
}

//
// Called by the event service to disconnect the supplier.
//
void
TEC_ProxyPushSupplier_impl::disconnect()
{
    {
	JTCSynchronized sync(*this);
	
	if(state_ == Destroyed)
	    return;
	
	state_ = Destroyed;
    }
    
    //
    // Ask the event pusher to stop, wait for it to complete, if
    // running.
    //
    if(pusher_)
    {
        pusher_ -> halt();
        while(pusher_ -> isAlive())
        {
            try
            {
                pusher_ -> join();
            }
            catch(const JTCInterruptedException&)
            {
            }
        }
    }

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

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