📄 ec_observerstrategy.cpp

📁 这是广泛使用的通信开源项目,对于大容量,高并发的通讯要求完全能够胜任,他广泛可用于网络游戏医学图像网关的高qos要求.更详细的内容可阅读相应的材料
💻 CPP
📖 第 1 页 / 共 2 页
字号:
上一页 12
      ACE_ENV_ARG_DECL)
{
  Headers headers;

  TAO_EC_Accumulate_Supplier_Headers worker (headers);
  this->event_channel_->for_each_consumer (&worker
                                           ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  RtecEventChannelAdmin::DependencySet& dep = qos.dependencies;

  dep.length (ACE_static_cast (CORBA::ULong, headers.current_size () + 1));

  dep[0].event.header.type = ACE_ES_DISJUNCTION_DESIGNATOR;
  dep[0].event.header.source = ACE_static_cast (CORBA::ULong,
                                                headers.current_size ());
  dep[0].event.header.creation_time = ORBSVCS_Time::zero ();
  dep[0].rt_info = 0;

  CORBA::ULong count = 1;
  for (HeadersIterator i = headers.begin (); i != headers.end (); ++i)
    {
      dep[count++].event.header = (*i).key ();
    }
}

void
TAO_EC_Basic_ObserverStrategy::fill_qos (
      RtecEventChannelAdmin::SupplierQOS &qos
      ACE_ENV_ARG_DECL)
{
  Headers headers;

  TAO_EC_Accumulate_Consumer_Headers worker (headers);
  this->event_channel_->for_each_supplier (&worker
                                           ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  qos.publications.length (ACE_static_cast (CORBA::ULong,
                                            headers.current_size ()));

  CORBA::ULong count = 0;
  for (HeadersIterator i = headers.begin (); i != headers.end (); ++i)
    {
      qos.publications[count++].event.header = (*i).key ();
    }
}


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

TAO_EC_Reactive_ObserverStrategy::~TAO_EC_Reactive_ObserverStrategy (void)
{
}

void
TAO_EC_Reactive_ObserverStrategy::supplier_qos_update (
                                        TAO_EC_ProxyPushConsumer *consumer
                                        ACE_ENV_ARG_DECL)
{
  if (consumer->publications ().is_gateway)
    return;

  RtecEventChannelAdmin::SupplierQOS s_qos;
  this->fill_qos (s_qos ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  Observer_Map copy;
  this->create_observer_map (copy ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  Observer_Map_Iterator end = copy.end ();
  for (Observer_Map_Iterator i  = copy.begin ();
       i != end;
       ++i)
    {
      Observer_Entry& entry = (*i).int_id_;
      ACE_TRY
        {
          entry.observer->update_supplier (s_qos ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK;
        }
      ACE_CATCH (CORBA::OBJECT_NOT_EXIST, ex)
        {
          // Exception occured while updating observer, so remove it from the
          // observer list
          this->observer_not_exists (entry ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK;
        }
      ACE_CATCH (CORBA::TRANSIENT, transient)
        {
          // Exception occured while updating observer, so remove it from the
          // observer list
          this->observer_not_exists (entry ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK;
        }
      ACE_CATCHANY
        {
          // Ignore all exceptions
        }
      ACE_ENDTRY;
    }
}

void
TAO_EC_Reactive_ObserverStrategy::consumer_qos_update (
                                        TAO_EC_ProxyPushSupplier *supplier
                                        ACE_ENV_ARG_DECL)
{
  if (supplier->subscriptions ().is_gateway)
    return;

  RtecEventChannelAdmin::ConsumerQOS c_qos;
  this->fill_qos (c_qos ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  Observer_Map copy;
  this->create_observer_map (copy ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  Observer_Map_Iterator end = copy.end ();
  for (Observer_Map_Iterator i  = copy.begin ();
       i != end;
       ++i)
    {
      Observer_Entry& entry = (*i).int_id_;
      ACE_TRY
        {
          entry.observer->update_consumer (c_qos ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK;
        }
      ACE_CATCH (CORBA::OBJECT_NOT_EXIST, ex)
        {
          // Exception occured while updating observer, so remove it from the
          // observer list
          this->observer_not_exists (entry ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK;
        }
      ACE_CATCH (CORBA::TRANSIENT, transient)
        {
          // Exception occured while updating observer, so remove it from the
          // observer list
          this->observer_not_exists (entry ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK;
        }
      ACE_CATCHANY
        {
          // Ignore all exceptions
        }
      ACE_ENDTRY;
    }
}

int
TAO_EC_Reactive_ObserverStrategy::create_observer_map (Observer_Map &map
                                                       ACE_ENV_ARG_DECL)
{
  ACE_GUARD_THROW_EX (ACE_Lock, ace_mon, *this->lock_,
                 RtecEventChannelAdmin::EventChannel::SYNCHRONIZATION_ERROR());
  ACE_CHECK_RETURN (0);

  Observer_Map_Iterator end = this->observers_.end ();
  for (Observer_Map_Iterator i  = this->observers_.begin ();
       i != end;
       ++i)
    {
      Observer_Entry& entry = (*i).int_id_;
      Observer_Entry copy (entry.handle,
                            RtecEventChannelAdmin::Observer::_duplicate (entry.observer.in ()));
      if (map.bind (copy.handle, copy) == -1)
      {
        map.unbind_all();
        return 0;
      }
    }

  return map.current_size();
}

void
TAO_EC_Reactive_ObserverStrategy::observer_not_exists (Observer_Entry& observer
                                                       ACE_ENV_ARG_DECL)
{
  ACE_TRY
    {
      this->remove_observer(observer.handle ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;
    }
  ACE_CATCHANY
    {
      // Ignore exceptions
    }
  ACE_ENDTRY;
}

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

void
TAO_EC_Accumulate_Supplier_Headers::work (TAO_EC_ProxyPushSupplier *supplier
                                          ACE_ENV_ARG_DECL_NOT_USED)
{
  const RtecEventChannelAdmin::ConsumerQOS& sub =
    supplier->subscriptions ();
  if (sub.is_gateway)
    return;
  for (CORBA::ULong j = 0; j < sub.dependencies.length (); ++j)
    {
      const RtecEventComm::Event& event =
        sub.dependencies[j].event;
      RtecEventComm::EventType type = event.header.type;

      if (0 < type && type < ACE_ES_EVENT_UNDEFINED)
        continue;
      this->headers_.insert (event.header, 1);
    }
}

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

void
TAO_EC_Accumulate_Consumer_Headers::work (TAO_EC_ProxyPushConsumer *consumer
                                          ACE_ENV_ARG_DECL_NOT_USED)
{
  const RtecEventChannelAdmin::SupplierQOS& pub =
    consumer->publications ();
  if (pub.is_gateway)
    return;
  for (CORBA::ULong j = 0; j < pub.publications.length (); ++j)
    {
      const RtecEventComm::Event& event =
        pub.publications[j].event;
      RtecEventComm::EventType type = event.header.type;

      if (0 < type && type < ACE_ES_EVENT_UNDEFINED)
        continue;
      this->headers_.insert (event.header, 1);
    }
}

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

#if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION)

template class ACE_Map_Manager<RtecEventChannelAdmin::Observer_Handle,TAO_EC_Basic_ObserverStrategy::Observer_Entry,ACE_Null_Mutex>;
template class ACE_Map_Iterator<RtecEventChannelAdmin::Observer_Handle,TAO_EC_Basic_ObserverStrategy::Observer_Entry,ACE_Null_Mutex>;
template class ACE_Map_Iterator_Base<RtecEventChannelAdmin::Observer_Handle,TAO_EC_Basic_ObserverStrategy::Observer_Entry,ACE_Null_Mutex>;
template class ACE_Map_Reverse_Iterator<RtecEventChannelAdmin::Observer_Handle,TAO_EC_Basic_ObserverStrategy::Observer_Entry,ACE_Null_Mutex>;
template class ACE_Map_Entry<RtecEventChannelAdmin::Observer_Handle,TAO_EC_Basic_ObserverStrategy::Observer_Entry>;

template class ACE_RB_Tree<RtecEventComm::EventHeader,int,TAO_EC_Basic_ObserverStrategy::Header_Compare,ACE_Null_Mutex>;
template class ACE_RB_Tree_Iterator_Base<RtecEventComm::EventHeader,int,TAO_EC_Basic_ObserverStrategy::Header_Compare,ACE_Null_Mutex>;
template class ACE_RB_Tree_Iterator<RtecEventComm::EventHeader,int,TAO_EC_Basic_ObserverStrategy::Header_Compare,ACE_Null_Mutex>;
template class ACE_RB_Tree_Reverse_Iterator<RtecEventComm::EventHeader,int,TAO_EC_Basic_ObserverStrategy::Header_Compare,ACE_Null_Mutex>;
template class ACE_RB_Tree_Node<RtecEventComm::EventHeader,int>;

template class ACE_Auto_Basic_Array_Ptr<RtecEventChannelAdmin::Observer_var>;

#elif defined(ACE_HAS_TEMPLATE_INSTANTIATION_PRAGMA)

#pragma instantiate ACE_Map_Manager<RtecEventChannelAdmin::Observer_Handle,TAO_EC_Basic_ObserverStrategy::Observer_Entry,ACE_Null_Mutex>
#pragma instantiate ACE_Map_Iterator<RtecEventChannelAdmin::Observer_Handle,TAO_EC_Basic_ObserverStrategy::Observer_Entry,ACE_Null_Mutex>
#pragma instantiate ACE_Map_Iterator_Base<RtecEventChannelAdmin::Observer_Handle,TAO_EC_Basic_ObserverStrategy::Observer_Entry,ACE_Null_Mutex>
#pragma instantiate ACE_Map_Reverse_Iterator<RtecEventChannelAdmin::Observer_Handle,TAO_EC_Basic_ObserverStrategy::Observer_Entry,ACE_Null_Mutex>
#pragma instantiate ACE_Map_Entry<RtecEventChannelAdmin::Observer_Handle,TAO_EC_Basic_ObserverStrategy::Observer_Entry>

#pragma instantiate ACE_RB_Tree<RtecEventComm::EventHeader,int,TAO_EC_Basic_ObserverStrategy::Header_Compare,ACE_Null_Mutex>
#pragma instantiate ACE_RB_Tree_Iterator_Base<RtecEventComm::EventHeader,int,TAO_EC_Basic_ObserverStrategy::Header_Compare,ACE_Null_Mutex>
#pragma instantiate ACE_RB_Tree_Iterator<RtecEventComm::EventHeader,int,TAO_EC_Basic_ObserverStrategy::Header_Compare,ACE_Null_Mutex>
#pragma instantiate ACE_RB_Tree_Reverse_Iterator<RtecEventComm::EventHeader,int,TAO_EC_Basic_ObserverStrategy::Header_Compare,ACE_Null_Mutex>
#pragma instantiate ACE_RB_Tree_Node<RtecEventComm::EventHeader,int>

#pragma instantiate ACE_Auto_Basic_Array_Ptr<RtecEventChannelAdmin::Observer_var>

#endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION */
上一页 12
⌨️ 快捷键说明

复制代码 Ctrl + C
搜索代码 Ctrl + F
全屏模式 F11
切换主题 Ctrl + Shift + D
显示快捷键 ?
增大字号 Ctrl + =
减小字号 Ctrl + -