#include "orbsvcs/Event/EC_Per_Supplier_Filter.h"
#include "orbsvcs/Event/EC_Event_Channel_Base.h"
#include "orbsvcs/Event/EC_ProxySupplier.h"
#include "orbsvcs/Event/EC_ProxyConsumer.h"
#include "orbsvcs/Event/EC_Scheduling_Strategy.h"
#include "orbsvcs/Event/EC_QOS_Info.h"

#include "orbsvcs/ESF/ESF_Proxy_Collection.h"

#include "orbsvcs/Event_Service_Constants.h"

#if ! defined (__ACE_INLINE__)
#include "orbsvcs/Event/EC_Per_Supplier_Filter.inl"
#endif /* __ACE_INLINE__ */



TAO_BEGIN_VERSIONED_NAMESPACE_DECL

TAO_EC_Per_Supplier_Filter::
    TAO_EC_Per_Supplier_Filter (TAO_EC_Event_Channel_Base* ec)
  :  event_channel_ (ec),
     consumer_ (0),
     refcnt_ (1)
{
  this->event_channel_->create_proxy_collection (this->collection_);
}

TAO_EC_Per_Supplier_Filter::~TAO_EC_Per_Supplier_Filter (void)
{
  this->event_channel_->destroy_proxy_collection (this->collection_);
  this->collection_ = 0;
}

void
TAO_EC_Per_Supplier_Filter::bind (TAO_EC_ProxyPushConsumer* consumer)
{
  ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);

  if (this->consumer_ == 0)
    {
      this->consumer_ = consumer;
    }
}

void
TAO_EC_Per_Supplier_Filter::unbind (TAO_EC_ProxyPushConsumer* consumer)
{
  ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);

  if (this->consumer_ == 0 || this->consumer_ != consumer)
    return;

  this->consumer_ = 0;

  try
    {
      this->shutdown ();
    }
  catch (const CORBA::Exception&)
    {
      // @@ Ignore exceptions
    }
}

void
TAO_EC_Per_Supplier_Filter::connected (TAO_EC_ProxyPushSupplier* supplier)
{
  ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);

  if (this->consumer_ != 0)
    {
      const RtecEventChannelAdmin::SupplierQOS& pub =
        this->consumer_->publications_i ();

      for (CORBA::ULong j = 0; j < pub.publications.length (); ++j)
        {
          const RtecEventComm::Event& event =
            pub.publications[j].event;

#if TAO_EC_ENABLE_DEBUG_MESSAGES
          ORBSVCS_DEBUG ((LM_DEBUG, "Connecting consumer <%x> to <%x>, "
                      "trying event <%d:%d>  ",
                      supplier, this,
                      event.header.source, event.header.type));
#endif /* TAO_EC_ENABLED_DEBUG_MESSAGES */
          if (supplier->can_match (event.header))
            {
#if TAO_EC_ENABLE_DEBUG_MESSAGES
              ORBSVCS_DEBUG ((LM_DEBUG, "  matched\n"));
#endif /* TAO_EC_ENABLED_DEBUG_MESSAGES */
              this->collection_->connected (supplier);
              return;
            }
#if TAO_EC_ENABLE_DEBUG_MESSAGES
          ORBSVCS_DEBUG ((LM_DEBUG, "  not matched\n"));
#endif /* TAO_EC_ENABLED_DEBUG_MESSAGES */
        }
    }
}

void
TAO_EC_Per_Supplier_Filter::reconnected (TAO_EC_ProxyPushSupplier* supplier)
{
  ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);

  if (this->consumer_ != 0)
    {
      const RtecEventChannelAdmin::SupplierQOS& pub =
        this->consumer_->publications_i ();

      for (CORBA::ULong j = 0; j < pub.publications.length (); ++j)
        {
          const RtecEventComm::Event& event =
            pub.publications[j].event;

          //      ORBSVCS_DEBUG ((LM_DEBUG, "Trying %d:%d in %x\n",
          //                  event.header.source, event.header.type,
          //                  this));
          if (supplier->can_match (event.header))
            {
              //          ORBSVCS_DEBUG ((LM_DEBUG, "  matched %x\n", supplier));
              this->collection_->connected (supplier);
              return;
            }
        }
      this->collection_->disconnected (supplier);
    }
}

void
TAO_EC_Per_Supplier_Filter::disconnected (TAO_EC_ProxyPushSupplier* supplier)
{
  this->collection_->disconnected (supplier);
}

void
TAO_EC_Per_Supplier_Filter::shutdown (void)
{
  this->collection_->shutdown ();
}

void
TAO_EC_Per_Supplier_Filter::push (const RtecEventComm::EventSet& event,
                                  TAO_EC_ProxyPushConsumer *consumer)
{
  TAO_EC_Scheduling_Strategy* scheduling_strategy =
    this->event_channel_->scheduling_strategy ();
  scheduling_strategy->schedule_event (event,
                                       consumer,
                                       this);
}

void
TAO_EC_Per_Supplier_Filter::push_scheduled_event (RtecEventComm::EventSet &event,
                                                  const TAO_EC_QOS_Info &event_info)
{
  TAO_EC_Filter_Worker worker (event, event_info);
  this->collection_->for_each (&worker);
}

CORBA::ULong
TAO_EC_Per_Supplier_Filter::_incr_refcnt (void)
{
  ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ace_mon, this->lock_, 0);

  this->refcnt_++;
  return this->refcnt_;
}

CORBA::ULong
TAO_EC_Per_Supplier_Filter::_decr_refcnt (void)
{
  {
    ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ace_mon, this->lock_, 0);

    this->refcnt_--;
    if (this->refcnt_ != 0)
      return this->refcnt_;
  }
  this->event_channel_->supplier_filter_builder ()->destroy (this);
  return 0;
}

// ****************************************************************

TAO_EC_Supplier_Filter*
TAO_EC_Per_Supplier_Filter_Builder::create (
    RtecEventChannelAdmin::SupplierQOS&)
{
  return new TAO_EC_Per_Supplier_Filter (this->event_channel_);
}

void
TAO_EC_Per_Supplier_Filter_Builder::destroy (
    TAO_EC_Supplier_Filter* x)
{
  delete x;
}

TAO_END_VERSIONED_NAMESPACE_DECL
