// **********************************************************************
//
// 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 <ProxySupplier_impl.h>
#include <ProxyState.h>
#include <EC_ProxyPullSupplier_impl.h>
#include <EC_ConsumerAdmin_impl.h>
#include <PropagatorBaseThread.h>
#include <EventUtil.h>
#include <Event_impl.h>
#include <QoS.h>

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

// ----------------------------------------------------------------------
// EC_ProxyPullSupplier_impl private member implementation
// ----------------------------------------------------------------------

Event_impl*
EC_ProxyPullSupplier_impl::getFirstEvent()
{
    assert(events_.length() > 0);
    Event_impl* event = events_[events_.length() - 1];
    events_.length(events_.length() - 1);
    return event;
}

// ----------------------------------------------------------------------
// EC_ProxyPullSupplier_impl constructor/destructor
// ----------------------------------------------------------------------

EC_ProxyPullSupplier_impl::EC_ProxyPullSupplier_impl(
    CORBA_ORB_ptr orb,
    EC_ConsumerAdmin_impl* admin,
    CORBA_ULong id)
    : orb_(CORBA_ORB::_duplicate(orb)),
      admin_(admin),
      state_(Disconnected),
      id_(id)
{
    //
    // Resolve the QoS params
    //
    OBProperties* properties = OBProperties::instance();
    maxEvents_ = atoi(properties -> getProperty(EventQoS::MAX_EVENTS));

#ifdef OB_TRACE
    if(OBMessageViewer::instance() -> getTraceLevel() >= 2)
    {
        CORBA_String_var desc = EventUtil::getConnectionAddrDescription(orb_);
        CORBA_String_var msg = CORBA_string_dup("Create ProxyPullSupplier(");
        msg += id_;
        msg += ") from ";
        msg += desc;
        OBMessageViewer::instance() -> trace(2, msg);
    }
#endif
}

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

    //
    // Discard each event in the queue
    //
    for(CORBA_ULong i = 0; i < events_.length(); ++i)
        events_[i] -> discard();
}

// ----------------------------------------------------------------------
// EC_ProxyPullSupplier_impl public member implementation
// ----------------------------------------------------------------------

void
EC_ProxyPullSupplier_impl::connect_pull_consumer(
    CosEventComm_PullConsumer_ptr consumer)
{
    JTCSynchronized sync(*this);
    
    if(state_ == Destroyed)
	throw CORBA_OBJECT_NOT_EXIST();

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

    if(!CORBA_is_nil(consumer))
	consumer_ = CosEventComm_PullConsumer::_duplicate(consumer);

    state_ = Connected;
}

CORBA_Any*
EC_ProxyPullSupplier_impl::pull()
{
    JTCSynchronized sync(*this);

    //
    // Block until there is an event, and we're still connected.
    //
    while(events_.length() == 0 && state_ == Connected)
    {
        while(true)
        {
            try
            {
                wait();
            }
            catch(const JTCInterruptedException&)
            {
                continue;
            }
            break;
        }
    }
    
    //
    // If we're not connected, then raise an exception.
    //
    if(state_ != Connected)
        throw CosEventComm_Disconnected();

    //
    // Get the first event, and return the data.
    //
    Event_impl* event = getFirstEvent();
    
    CORBA_Any* any = new CORBA_Any(event -> any());
    event -> discard();

    return any;
}

CORBA_Any*
EC_ProxyPullSupplier_impl::try_pull(CORBA_Boolean& hasEvent)
{
    JTCSynchronized sync(*this);

    //
    // If we're not connected, then raise an exception.
    //
    if(state_ != Connected)
        throw CosEventComm_Disconnected();

    //
    // Determine if we have an event.
    //
    hasEvent = events_.length() != 0;
    CORBA_Any* any;
    if(hasEvent)
    {
        //
        // Yes, then get the first event. Return the data.
        //
        Event_impl* event = getFirstEvent();
        any = new CORBA_Any(event -> any());
        event -> discard();
    }
    else
    {
        any = new CORBA_Any;
    }
    return any;
}

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

	//
	// Notify on state change.
	//
	notify();
    }

    admin_ -> removeProxy(this);

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

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

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

    //
    // Don't accumulate new events while the channel isn't connected.
    //
    if(state_ != Connected)
        return;

    //
    // If we queue more than maxEvents_ items then we throw oldest
    // items away first.
    //
    events_.insert(event);
    event -> acquire();
    
    if(events_.length() > maxEvents_)
    {
        Event_impl* e = events_[events_.length() - 1];
#ifdef OB_TRACE
        if(OBMessageViewer::instance() -> getTraceLevel() >= 2)
        {
            CORBA_String_var msg =
                CORBA_string_dup("ProxyPullSupplier(");
            msg += id_;
            msg += ") Discarding event (";
            msg += e -> eventId();
            msg += ')';
            OBMessageViewer::instance() -> trace(2, msg);
        }
#endif
        e -> discard();
        events_.length(events_.length() - 1);
    }

    //
    // Wake any waiting pull() events.
    //
    notify();
}

//
// Called by the event service to disconnect the supplier.
//
void
EC_ProxyPullSupplier_impl::disconnect()
{
    JTCSynchronized sync(*this);

    if(state_ == Destroyed)
        return;

    //
    // Mark the servant as destroyed, wake any waiting threads.
    //
    state_ = Destroyed;
    notify();
    
    //
    // Disconnect the servant from the ORB.
    //
    orb_ -> disconnect(this);
}

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