ec_kokyu_filter_builder.cpp

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

// EC_Kokyu_Filter_Builder.cpp,v 1.6 2003/11/04 05:21:32 dhinton Exp

#include "EC_Kokyu_Filter.h"
#include "orbsvcs/Event_Service_Constants.h"
#include "EC_Kokyu_Filter_Builder.h"
#include "EC_Type_Filter.h"
#include "EC_Conjunction_Filter.h"
#include "EC_Disjunction_Filter.h"
#include "EC_Timeout_Filter.h"
#include "EC_Event_Channel_Base.h"
#include "ace/OS_NS_stdio.h"

#if ! defined (__ACE_INLINE__)
#include "EC_Kokyu_Filter_Builder.i"
#endif /* __ACE_INLINE__ */

ACE_RCSID(Event, EC_Kokyu_Filter_Builder, "EC_Kokyu_Filter_Builder.cpp,v 1.6 2003/11/04 05:21:32 dhinton Exp")

const char* designator (long dsgn)
{
  switch(dsgn)
    {
    case ACE_ES_GLOBAL_DESIGNATOR: return "GLOBAL";
    case ACE_ES_CONJUNCTION_DESIGNATOR: return "CONJ";
    case ACE_ES_DISJUNCTION_DESIGNATOR: return "DISJ";
    case ACE_ES_NEGATION_DESIGNATOR: return "NEG";
    case ACE_ES_LOGICAL_AND_DESIGNATOR: return "LOG_AND";
    case ACE_ES_BITMASK_DESIGNATOR: return "BITMASK";
    case ACE_ES_MASKED_TYPE_DESIGNATOR: return "MASKED_TYPE";
    case ACE_ES_NULL_DESIGNATOR: return "NULL";
    }

  return "---";
}

TAO_EC_Kokyu_Filter_Builder::~TAO_EC_Kokyu_Filter_Builder (void)
{
}

TAO_EC_Filter*
TAO_EC_Kokyu_Filter_Builder::build (
    TAO_EC_ProxyPushSupplier *supplier,
    RtecEventChannelAdmin::ConsumerQOS& qos
    ACE_ENV_ARG_DECL) const
{
  CORBA::ULong i=0,found=0;
  CORBA::ULong pos = 0;
  CORBA::Long npos = -1;
  int establish_final_consumer_dependency=0;

  CORBA::Object_var tmp =
    this->event_channel_->scheduler ();

  RtecScheduler::Scheduler_var scheduler =
    RtecScheduler::Scheduler::_narrow (tmp.in () ACE_ENV_ARG_PARAMETER);
  ACE_CHECK_RETURN (0);

#ifdef EC_KOKYU_LOGGING
  for (i=0; i<qos.dependencies.length (); ++i)
    {                  
      ACE_DEBUG ((LM_DEBUG,
                  "consumerqos[%d] event.header.type = %s,"
                  "rt_info = %d\n",
                  i,
                  designator (qos.dependencies[i].event.header.type),
                  qos.dependencies[i].rt_info));
    }
#endif

  //find the first entry which is not a designator. We are going to
  //assume that this entry will have the rt_info of the connecting
  //consumer (ProxyPushSupplier), which is passed into this function.
  for (i=0; !found && i<qos.dependencies.length (); ++i)
    {                  
      switch (qos.dependencies[i].event.header.type)
        {
        case ACE_ES_CONJUNCTION_DESIGNATOR:
        case ACE_ES_DISJUNCTION_DESIGNATOR:
        case ACE_ES_NEGATION_DESIGNATOR:
        case ACE_ES_LOGICAL_AND_DESIGNATOR:
        case ACE_ES_BITMASK_DESIGNATOR:
        case ACE_ES_MASKED_TYPE_DESIGNATOR:
        case ACE_ES_NULL_DESIGNATOR:
          establish_final_consumer_dependency = 1;
          continue;

        case ACE_ES_GLOBAL_DESIGNATOR:
        case ACE_ES_EVENT_TIMEOUT:
        case ACE_ES_EVENT_INTERVAL_TIMEOUT:
        case ACE_ES_EVENT_DEADLINE_TIMEOUT:
          continue;
          
        default:
          npos = i;
          found = 1;
          break;
        }
    }

  ACE_CString final_consumer_rep_name; 
  RtecScheduler::handle_t h_final_consumer_rt_info = 0;
  RtecScheduler::handle_t h_final_consumer_rep_rt_info = 0;

#ifdef EC_KOKYU_LOGGING
  ACE_DEBUG ((LM_DEBUG, "consumer rt_info found in consumerqos[%d] \n", npos));
#endif

  if (npos >= 0 && establish_final_consumer_dependency == 1)
    {
      //Hopefully this will have the final consumer's rt_info
      h_final_consumer_rt_info = qos.dependencies[npos].rt_info;

#ifdef EC_KOKYU_LOGGING
      ACE_DEBUG ((LM_DEBUG, "about to get rt_info =  %d\n",
                  h_final_consumer_rep_rt_info));
#endif

      RtecScheduler::RT_Info_var final_consumer_rt_info =
        scheduler->get ( h_final_consumer_rt_info 
                         ACE_ENV_ARG_PARAMETER);

      final_consumer_rep_name = final_consumer_rt_info->entry_point.in ();
      final_consumer_rep_name += "#rep";

#ifdef EC_KOKYU_LOGGING
      ACE_DEBUG ((LM_DEBUG, "about to create consumer rep %s\n",
                  final_consumer_rep_name.c_str ()));
#endif

      //create an rt_info corresponding to this rep.
      h_final_consumer_rep_rt_info =
        scheduler->create (final_consumer_rep_name.c_str () 
                           ACE_ENV_ARG_PARAMETER);
      ACE_CHECK_RETURN (0);
#ifdef EC_KOKYU_LOGGING
      ACE_DEBUG ((LM_DEBUG, "consumer rep created\n"));
#endif

    }

  //We are passing the final consumer as the parent. The final
  //consumer is the one which is connecting to the ProxyPushSupplier
  //passed in to this function.

  TAO_EC_Filter* filter = 
    this->recursive_build (supplier, qos, pos,
                           scheduler.in (),
                           h_final_consumer_rep_rt_info  //parent_info
                           ACE_ENV_ARG_PARAMETER);

#ifdef EC_KOKYU_LOGGING
  ACE_DEBUG ((LM_DEBUG, 
              "Filter_Builder::Verifying whether root filter"
              " dependency can be established\n"));
#endif

  if (npos >= 0 && establish_final_consumer_dependency == 1)
    {
#ifdef EC_KOKYU_LOGGING
      ACE_DEBUG ((LM_DEBUG, 
                  "Filter_Builder::root filter dependency "
                  "can be established\n"));
#endif
      TAO_EC_Kokyu_Filter* kokyu_filter =
        ACE_dynamic_cast(TAO_EC_Kokyu_Filter*, filter);
  
      //add the dependency between the root in the filter hierarchy and
      //the final consumer
      TAO_EC_QOS_Info qos_info;
      kokyu_filter->get_qos_info (qos_info ACE_ENV_ARG_PARAMETER);

      scheduler->add_dependency (h_final_consumer_rt_info, 
                                 qos_info.rt_info,
                                 1,
                                 RtecBase::ONE_WAY_CALL
                                 ACE_ENV_ARG_PARAMETER);
      ACE_CHECK_RETURN (0);
    }
  return filter;
}

TAO_EC_Filter*
TAO_EC_Kokyu_Filter_Builder::recursive_build (
    TAO_EC_ProxyPushSupplier *supplier,
    RtecEventChannelAdmin::ConsumerQOS& qos,
    CORBA::ULong& pos,
    RtecScheduler::Scheduler_ptr scheduler,
    RtecScheduler::handle_t parent_info
    ACE_ENV_ARG_DECL) const
{
  const RtecEventComm::Event& e = qos.dependencies[pos].event;

#ifdef EC_KOKYU_LOGGING
  ACE_DEBUG ((LM_DEBUG, "Filter_Builder::In recursive build\n"));
#endif

  if (e.header.type == ACE_ES_CONJUNCTION_DESIGNATOR)
    {
#ifdef EC_KOKYU_LOGGING
      ACE_DEBUG ((LM_DEBUG, "Filter_Builder::Conjuction designator\n"));
#endif
      CORBA::ULong npos = pos;
      ACE_CString name;
      this->recursive_name (qos, npos,
                            scheduler, name
                             ACE_ENV_ARG_PARAMETER);
      ACE_CHECK_RETURN (0);

      pos++; // Consume the designator

      CORBA::ULong n = this->count_children (qos, pos);

      RtecBase::handle_t conj_rt_info = parent_info;

      TAO_EC_Filter** children;
      ACE_NEW_RETURN (children, TAO_EC_Filter*[n], 0);
      for (CORBA::ULong i = 0; i != n; ++i)
        {
          children[i] = this->recursive_build (supplier, qos, pos,
                                               scheduler,
                                               conj_rt_info
                                               ACE_ENV_ARG_PARAMETER);
          ACE_CHECK_RETURN (0);
        }

      TAO_EC_Kokyu_Filter *filter;
      ACE_NEW_RETURN (filter,
                      TAO_EC_Kokyu_Filter (name.c_str (),
                                           conj_rt_info,
                                           scheduler,
                                           new TAO_EC_Conjunction_Filter(children,
                                                                         n),
                                           conj_rt_info,
                                           conj_rt_info,
                                           RtecScheduler::CONJUNCTION),
                      0);
      TAO_EC_QOS_Info qos_info;
      filter->get_qos_info (qos_info ACE_ENV_ARG_PARAMETER);
      // @@
      ACE_CHECK_RETURN (0);
      return filter;
    }

  else if (e.header.type == ACE_ES_DISJUNCTION_DESIGNATOR)
    {
#ifdef EC_KOKYU_LOGGING
      ACE_DEBUG ((LM_DEBUG, "Filter_Builder::Disjunction designator\n"));
#endif
      CORBA::ULong npos = pos;
      ACE_CString name;
      this->recursive_name (qos, npos,
                            scheduler, name
                             ACE_ENV_ARG_PARAMETER);
      ACE_CHECK_RETURN (0);