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

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

#include <OB/CosEventChannelAdmin.h>

#include <PropagatorBaseThread.h>
#include <PusherBaseThread.h>
#include <Event_impl.h>

// ----------------------------------------------------------------------
// Template instantiations
// ----------------------------------------------------------------------

#ifndef HAVE_NO_EXPLICIT_TEMPLATES
template class JTCHandleT<PusherBaseThread>;
#else
#ifdef HAVE_PRAGMA_DEFINE
#pragma define(JTCHandleT<PusherBaseThread>)
#endif
#endif

// ----------------------------------------------------------------------
// PusherBaseThread private member implementation
// ----------------------------------------------------------------------

//
// Wait until we are terminated or for the next event to
// arrive. Return the event if available, 0 if not.
//
Event_impl*
PusherBaseThread::waitEvent()
{
    JTCSynchronized sync(*this);
    
    while(!done_ && events_.length() == 0)
    {
        try
        {
            wait();
        }
        catch(const JTCInterruptedException&)
        {
        }
    }
    
    Event_impl* event;
    if(!done_)
    {
        assert(events_.length() > 0);
        event = events_[events_.length() - 1];

        //
        // Acquire a reference to the event. This is so that if the
        // event is dropped from the event queue, then this event is
        // still valid.
        //
        event -> acquire();
    }
    else
    {
        event = 0;
    }
    return event;
}

void
PusherBaseThread::removeEvent(Event_impl* event)
{
    JTCSynchronized sync(*this);
    
    //
    // We know that either this event is either the last event, or not
    // in the event list at all.
    //
    if(events_[events_.length() - 1] == event)
    {
        events_.length(events_.length() - 1);
        event -> discard();
    }
}

// ----------------------------------------------------------------------
// PusherBaseThread constructor/destructor
// ----------------------------------------------------------------------

PusherBaseThread::PusherBaseThread(
    CosEventComm_PushConsumer_ptr consumer,
    CosEventChannelAdmin_ProxyPushSupplier_ptr proxy,
    CORBA_ULong id,
    CORBA_ULong maxEvents,
    CORBA_ULong maxRetries,
    CORBA_ULong retryTimeout,
    CORBA_Double retryMultiplier)
    
    : PropagatorBaseThread(id, maxRetries, retryTimeout, retryMultiplier),
      maxEvents_(maxEvents),
      consumer_(CosEventComm_PushConsumer::_duplicate(consumer)),
      proxy_(CosEventChannelAdmin_ProxyPushSupplier::_duplicate(proxy))
{
}

PusherBaseThread::~PusherBaseThread()
{
#ifdef OB_TRACE
    OBMessageViewer::instance() -> trace(2, "Destroy PusherBaseThread");
#endif
    
    //
    // Discard each event in the queue
    //
    for(CORBA_ULong i = 0; i < events_.length(); ++i)
        events_[i] -> discard();
}

// ----------------------------------------------------------------------
// PusherBaseThread public member implementation
// ----------------------------------------------------------------------

//
// Notify the pusher that there is a new event
//
void
PusherBaseThread::newEvent(Event_impl* event)
{
    JTCSynchronized sync(*this);
    
    events_.insert(event);
    event -> acquire();

    //
    // Too many events in the queue? Drop the last event.
    //
    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("PusherBaseThread(");
            msg += id_;
            msg += ") Discarding event (";
            msg += e -> eventId();
            msg += ')';
            OBMessageViewer::instance() -> trace(2, msg);
        }
#endif

        //
        // Discard the event.
        //
        e -> discard();

        //
        // The queue is now one event less.
        //
        events_.length(events_.length() - 1);
    }
    
    notify();
}

//
// Run the pusher. Wait for an event, then ask the supplier to
// push the first event.
//
void
PusherBaseThread::run()
{
    Event_impl* event;
    
    while((event = waitEvent()) != 0)
    {
        switch(pushEvent(event))
        {
        case Success:
            resetRetry();
            removeEvent(event);
            event -> discard();
            break;

        case Retry:
            //
            // First, discard the event. Set the event to 0 so we can
            // fall through to the Failed case if we're not to retry
            // again.
            //
            event -> discard();
            event = 0;
            
            //
            // If retry returns true, then we should send the event at
            // the front of the event queue.
            //
            if(retry())
                continue;
            
#ifdef OB_TRACE
            if(OBMessageViewer::instance() -> getTraceLevel() >= 2)
            {
                CORBA_String_var msg =
                    CORBA_string_dup("Retries exceeded: disconnect ");
                msg += id_;
                OBMessageViewer::instance() -> trace(2, msg);
            }
#endif
            //
            // fall through
            //
            
        case Failed:
            if(event != 0)
                event -> discard();
            try
            {
                proxy_ -> disconnect_push_supplier();
                
            }
            catch(...)
            {
            }
            break;
        }
    }
    
    //
    // We're no longer interested in the ProxyPushSupplier. This is
    // needed to break the circular reference.
    //
    proxy_ = CosEventChannelAdmin_ProxyPushSupplier::_nil();

    //
    // The consumer is no longer of any interest either. We need to
    // clear this reference also.
    //
    consumer_ = CosEventComm_PushConsumer::_nil();
}
