ec_proxysupplier.cpp

这是广泛使用的通信开源项目,对于大容量,高并发的通讯要求完全能够胜任,他广泛可用于网络游戏医学图像网关的高qos要求.更详细的内容可阅读相应的材料

    RtecEventComm::PushConsumer::_duplicate (this->consumer_.in ());

  this->pre_dispatch_hook (ACE_const_cast (RtecEventComm::EventSet&, event)
                           ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  {
    // We have to release the lock to avoid dead-locks.
    TAO_EC_Unlock reverse_lock (*this->lock_);

    ACE_GUARD_THROW_EX (TAO_EC_Unlock, ace_mon, reverse_lock,
                        RtecEventChannelAdmin::EventChannel::SYNCHRONIZATION_ERROR ());
    ACE_CHECK;

    this->event_channel_->dispatching ()->push (this,
                                                consumer.in (),
                                                event,
                                                qos_info
                                                 ACE_ENV_ARG_PARAMETER);
    ACE_CHECK;
  }

  if (this->child_ != 0)
    this->child_->clear ();
}

void
TAO_EC_ProxyPushSupplier::pre_dispatch_hook (RtecEventComm::EventSet&
                                             ACE_ENV_ARG_DECL_NOT_USED)
{
}

void
TAO_EC_ProxyPushSupplier::push_nocopy (RtecEventComm::EventSet& event,
                                       TAO_EC_QOS_Info& qos_info
                                       ACE_ENV_ARG_DECL)
{
  // The mutex is already held by the caller (usually the filter()
  // method)
  if (this->is_connected_i () == 0)
    return; // TAO_THROW (RtecEventComm::Disconnected ());????

  if (this->suspended_ != 0)
    return;

  TAO_ESF_RefCount_Guard<CORBA::ULong> ace_mon (this->refcount_);
  // The guard will decrement the reference count, notice that the
  // reference count can become 0, but this is not the right spot to
  // check for that and destroy the object.
  // If we did so then we would destroy the object, and consequently
  // the mutex, but the mutex is used later when the stack unwinds and
  // the filter() method tries to destroy the mutex (that originally
  // acquired the mutex in the first place).
  // So the correct thing to do is to just decrement the reference
  // count and let the filter() method do the destruction.

  RtecEventComm::PushConsumer_var consumer =
    RtecEventComm::PushConsumer::_duplicate (this->consumer_.in ());

  this->pre_dispatch_hook (event ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  {
    TAO_EC_Unlock reverse_lock (*this->lock_);

    ACE_GUARD_THROW_EX (TAO_EC_Unlock, ace_mon, reverse_lock,
                        RtecEventChannelAdmin::EventChannel::SYNCHRONIZATION_ERROR ());
    ACE_CHECK;

    this->event_channel_->dispatching ()->push_nocopy (this,
                                                       consumer.in (),
                                                       event,
                                                       qos_info
                                                        ACE_ENV_ARG_PARAMETER);
    ACE_CHECK;
  }

  if (this->child_ != 0)
    this->child_->clear ();
}

void
TAO_EC_ProxyPushSupplier::push_to_consumer (
    RtecEventComm::PushConsumer_ptr consumer,
    const RtecEventComm::EventSet& event
    ACE_ENV_ARG_DECL)
{
  {
    ACE_GUARD_THROW_EX (
            ACE_Lock, ace_mon, *this->lock_,
            RtecEventChannelAdmin::EventChannel::SYNCHRONIZATION_ERROR ());
    ACE_CHECK;

    if (this->is_connected_i () == 0)
      return; // ACE_THROW (RtecEventComm::Disconnected ());????

    if (this->suspended_ != 0)
      return;
  }

  ACE_TRY
    {
      consumer->push (event ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;
    }
  ACE_CATCH (CORBA::OBJECT_NOT_EXIST, not_used)
    {
      // Do not report errors for old consumers
      // NOTE: The comparison below is not completely correct, it
      // could be that the remote consumer and the local consumer are
      // in fact the same object, but represented by different objects
      // references.  Unfortunately this is not a good spot to invoke
      // _is_equivalent(), and that may not give us the desired answer
      // anyway.
      if (consumer == this->consumer_.in ())
        {
          TAO_EC_ConsumerControl *control =
            this->event_channel_->consumer_control ();

          control->consumer_not_exist (this ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK;
        }
    }
  ACE_CATCH (CORBA::SystemException, sysex)
    {
      // Do not report errors for old consumers
      // NOTE: The comparison below is not completely correct, it
      // could be that the remote consumer and the local consumer are
      // in fact the same object, but represented by different objects
      // references.  Unfortunately this is not a good spot to invoke
      // _is_equivalent(), and that may not give us the desired answer
      // anyway.
      if (consumer == this->consumer_.in ())
        {
          TAO_EC_ConsumerControl *control =
            this->event_channel_->consumer_control ();

          control->system_exception (this,
                                     sysex
                                      ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK;
        }
    }
  ACE_CATCHANY
    {
      // Shouldn't happen, but does not hurt
    }
  ACE_ENDTRY;
}

void
TAO_EC_ProxyPushSupplier::reactive_push_to_consumer (
    RtecEventComm::PushConsumer_ptr consumer,
    const RtecEventComm::EventSet& event
    ACE_ENV_ARG_DECL)
{
  ACE_TRY
    {
      consumer->push (event ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;
    }
  ACE_CATCH (CORBA::OBJECT_NOT_EXIST, not_used)
    {
      TAO_EC_ConsumerControl *control =
        this->event_channel_->consumer_control ();

      control->consumer_not_exist (this ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;
    }
  ACE_CATCH (CORBA::SystemException, sysex)
    {
      TAO_EC_ConsumerControl *control =
        this->event_channel_->consumer_control ();

      control->system_exception (this,
                                 sysex
                                  ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;
    }
  ACE_CATCHANY
    {
      // Shouldn't happen
    }
  ACE_ENDTRY;
}

CORBA::Boolean
TAO_EC_ProxyPushSupplier::consumer_non_existent (
      CORBA::Boolean_out disconnected
      ACE_ENV_ARG_DECL)
{
  CORBA::Object_var consumer;
  {
    ACE_GUARD_THROW_EX (
        ACE_Lock, ace_mon, *this->lock_,
        CORBA::INTERNAL ());
    ACE_CHECK_RETURN (0);

    disconnected = 0;
    if (this->is_connected_i () == 0)
      {
        disconnected = 1;
        return 0;
      }

    consumer = CORBA::Object::_duplicate (this->consumer_.in ());
  }

#if (TAO_HAS_MINIMUM_CORBA == 0)
  return consumer->_non_existent (ACE_ENV_SINGLE_ARG_PARAMETER);
#else
  return 0;
#endif /* TAO_HAS_MINIMUM_CORBA */
}

void
TAO_EC_ProxyPushSupplier::clear (void)
{
  ACE_GUARD (ACE_Lock, ace_mon, *this->lock_);

  this->child_->clear ();
}

CORBA::ULong
TAO_EC_ProxyPushSupplier::max_event_size (void) const
{
  ACE_GUARD_RETURN (ACE_Lock, ace_mon, *this->lock_, 0);

  return this->child_->max_event_size ();
}

int
TAO_EC_ProxyPushSupplier::can_match (
      const RtecEventComm::EventHeader &header) const
{
  ACE_GUARD_RETURN (ACE_Lock, ace_mon, *this->lock_, 0);

  if (this->is_connected_i () == 0)
    return 0;

  return this->child_->can_match (header);
}

int
TAO_EC_ProxyPushSupplier::add_dependencies (
      const RtecEventComm::EventHeader &header,
      const TAO_EC_QOS_Info &qos_info
      ACE_ENV_ARG_DECL)
{
  ACE_GUARD_THROW_EX (
          ACE_Lock, ace_mon, *this->lock_,
          RtecEventChannelAdmin::EventChannel::SYNCHRONIZATION_ERROR ());
  ACE_CHECK_RETURN (0);

  return this->child_->add_dependencies (header,
                                         qos_info
                                         ACE_ENV_ARG_PARAMETER);
}


#if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION)

template class TAO_ESF_RefCount_Guard<CORBA::ULong>;
template class TAO_ESF_Proxy_RefCount_Guard<TAO_EC_Event_Channel_Base,TAO_EC_ProxyPushSupplier>;

#elif defined(ACE_HAS_TEMPLATE_INSTANTIATION_PRAGMA)

#pragma instantiate TAO_ESF_RefCount_Guard<CORBA::ULong>
#pragma instantiate TAO_ESF_Proxy_RefCount_Guard<TAO_EC_Event_Channel_Base,TAO_EC_ProxyPushSupplier>

#endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION */