service.cpp

来自「这是广泛使用的通信开源项目,对于大容量,高并发的通讯要求完全能够胜任,他广泛可用」· C++ 代码 · 共 657 行 · 第 1/3 页

// Service.cpp,v 1.11 2003/11/04 05:21:31 dhinton Exp

#include "orbsvcs/Sched/Reconfig_Scheduler.h"
#include "orbsvcs/Runtime_Scheduler.h"
//#include "orbsvcs/Event/Module_Factory.h"
//#include "orbsvcs/Event/Event_Channel.h"
#include "orbsvcs/Event_Service_Constants.h"
#include "orbsvcs/Event_Utilities.h"
#include "orbsvcs/Scheduler_Factory.h"
#include "orbsvcs/Event/EC_Event_Channel.h"
#include "orbsvcs/Event/EC_Default_Factory.h"
#include "orbsvcs/Event/EC_Kokyu_Factory.h"
#include "Consumer.h"
#include "Supplier.h"


#include "ace/Get_Opt.h"
#include "ace/Sched_Params.h"
#include "ace/Auto_Ptr.h"
#include "ace/SString.h"
#include "ace/OS_NS_strings.h"
#include "ace/Thread.h"

ACE_RCSID(EC_Examples, Service, "Service.cpp,v 1.11 2003/11/04 05:21:31 dhinton Exp")

namespace
{
  int config_run = 0;
  ACE_CString sched_type ="rms";
}

inline RtecScheduler::Period_t time_val_to_period (const ACE_Time_Value &tv)
{
  //100s of nanoseconds
  return (tv.sec () * 1000000 + tv.usec ())*10;
}

int parse_args (int argc, char *argv[]);

typedef TAO_Reconfig_Scheduler<TAO_RMS_FAIR_Reconfig_Sched_Strategy, TAO_SYNCH_MUTEX> RECONFIG_RMS_SCHED_TYPE;

typedef TAO_Reconfig_Scheduler<TAO_MUF_FAIR_Reconfig_Sched_Strategy, TAO_SYNCH_MUTEX> RECONFIG_MUF_SCHED_TYPE;

int
main (int argc, char* argv[])
{
  //TAO_EC_Default_Factory::init_svcs ();

  TAO_EC_Kokyu_Factory::init_svcs ();


  ACE_DECLARE_NEW_CORBA_ENV;
  ACE_TRY
    {
      // ORB initialization boiler plate...
      CORBA::ORB_var orb =
        CORBA::ORB_init (argc, argv, "" ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      if (parse_args (argc, argv) == -1)
        {
          ACE_ERROR ((LM_ERROR,
                      "Usage: Service [-o IOR_file_name]\n"));
          return 1;
        }

      CORBA::Object_var object =
        orb->resolve_initial_references ("RootPOA" ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;
      PortableServer::POA_var poa =
        PortableServer::POA::_narrow (object.in () ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;
      PortableServer::POAManager_var poa_manager =
        poa->the_POAManager (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;
      poa_manager->activate (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

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

      // Create an scheduling service
      POA_RtecScheduler::Scheduler* sched_impl = 0;

      if (ACE_OS::strcasecmp(sched_type.c_str(), "rms") == 0)
        {
          ACE_DEBUG ((LM_DEBUG, "Creating RMS scheduler\n"));
          ACE_NEW_RETURN (sched_impl,
                          RECONFIG_RMS_SCHED_TYPE,
                          1);
        }
      else if (ACE_OS::strcasecmp(sched_type.c_str(), "muf") == 0)
        {
          ACE_DEBUG ((LM_DEBUG, "Creating MUF scheduler\n"));
          ACE_NEW_RETURN (sched_impl,
                          RECONFIG_MUF_SCHED_TYPE,
                          1);
        }

      RtecScheduler::Scheduler_var scheduler =
        sched_impl->_this (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

      // ****************************************************************
#if 0
      // Create an event channel implementation...
      TAO_Default_Module_Factory module_factory;
      ACE_EventChannel ec_impl (scheduler.in (),
                                1,
                                ACE_DEFAULT_EVENT_CHANNEL_TYPE,
                                &module_factory);

      RtecEventChannelAdmin::EventChannel_var event_channel =
        ec_impl._this (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;
#else
      TAO_EC_Event_Channel_Attributes attributes (poa.in (),
                                                  poa.in ());
      attributes.scheduler = scheduler.in (); // no need to dup

      TAO_EC_Event_Channel ec_impl (attributes);
      RtecEventChannelAdmin::EventChannel_var event_channel =
        ec_impl._this (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;
#endif
      // ****************************************************************

      // Create a consumer, intialize its RT_Info structures, and
      // connnect to the event channel....

      Consumer consumer_impl1, consumer_impl2;

      RtecScheduler::handle_t consumer1_rt_info =
        scheduler->create ("consumer1" ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      RtecScheduler::handle_t consumer2_rt_info =
        scheduler->create ("consumer2" ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      //consumer's rate will get propagated from the supplier.
      //so no need to specify a period here. Specifying
      //criticality is crucial since it propagates from
      //consumer to supplier.
      ACE_Time_Value tv (0,0);
      TimeBase::TimeT tmp;
      ORBSVCS_Time::Time_Value_to_TimeT (tmp, tv);
      scheduler->set (consumer1_rt_info,
                      RtecScheduler::VERY_LOW_CRITICALITY,
                      tmp, tmp, tmp,
                      time_val_to_period (tv),
                      RtecScheduler::VERY_LOW_IMPORTANCE,
                      tmp,
                      0,
                      RtecScheduler::OPERATION
                      ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      scheduler->set (consumer2_rt_info,
                      RtecScheduler::VERY_HIGH_CRITICALITY,
                      tmp, tmp, tmp,
                      time_val_to_period (tv),
                      RtecScheduler::VERY_HIGH_IMPORTANCE,
                      tmp,
                      0,
                      RtecScheduler::OPERATION
                      ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      ACE_ConsumerQOS_Factory consumer_qos1, consumer_qos2;
      //consumer_qos.start_disjunction_group ();
      // The types int the range [0,ACE_ES_EVENT_UNDEFINED) are
      // reserved for the EC...
      consumer_qos1.insert_type (ACE_ES_EVENT_UNDEFINED,
                                consumer1_rt_info);

      RtecEventChannelAdmin::ConsumerQOS qos =
        consumer_qos1.get_ConsumerQOS ();
/*
      for (int i=0;i<qos.dependencies.length (); ++i)
        {
          ACE_DEBUG ((LM_DEBUG,
                      "consumer_qos1[%d] event.header.type = %d, "
                      "consumer_qos1[%d] rt_info = %d, "
                      "consumer_qos1[%d] event.header.source = %d\n",
                      i,qos.dependencies[i].event.header.type,
                      i,qos.dependencies[i].rt_info,
                      i,qos.dependencies[i].event.header.source));
        }
*/

      consumer_qos2.insert_type (ACE_ES_EVENT_UNDEFINED + 1,
                                consumer2_rt_info);

      // The canonical protocol to connect to the EC
      RtecEventChannelAdmin::ConsumerAdmin_var consumer_admin =
        event_channel->for_consumers (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

      RtecEventChannelAdmin::ProxyPushSupplier_var supplier_proxy1 =
        consumer_admin->obtain_push_supplier (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

      RtecEventChannelAdmin::ProxyPushSupplier_var supplier_proxy2 =
        consumer_admin->obtain_push_supplier (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

      RtecEventComm::PushConsumer_var consumer1 =
        consumer_impl1._this (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

      RtecEventComm::PushConsumer_var consumer2 =
        consumer_impl2._this (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

      ACE_DEBUG ((LM_DEBUG, "connecting consumers\n"));
      ACE_DEBUG ((LM_DEBUG, "connecting consumer1\n"));
      supplier_proxy1->connect_push_consumer (consumer1.in (),
                                             consumer_qos1.get_ConsumerQOS ()
                                             ACE_ENV_ARG_PARAMETER);