dove_supplier.cpp

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

    event.header.type = ACE_ES_EVENT_NOTIFICATION;
    event.header.ttl = 1;
    ACE_hrtime_t creation_time = ACE_OS::gethrtime ();
    ORBSVCS_Time::hrtime_to_TimeT (event.header.creation_time, creation_time);
    event.header.ec_recv_time = ORBSVCS_Time::zero ();
    event.header.ec_send_time = ORBSVCS_Time::zero ();
    event.data.any_value = message;

    RtecEventComm::EventSet events;
    events.length (1);
    events[0] = event;

    // Now we invoke a RPC
    this->current_connection_params_->proxyPushConsumer_var_->push (events
                                                                    ACE_ENV_ARG_PARAMETER);
    ACE_TRY_CHECK;
  }
  ACE_CATCHANY
  {
    ACE_ERROR ((LM_ERROR,
                "DOVE_Supplier::notify: "
                "unexpected exception.\n"));
  }
  ACE_ENDTRY;
}


// Use the next connection in the list of established connections.

void
DOVE_Supplier::use_next_connection ()
{
  if (connection_count_ > 0)
    {
      current_connection_index_ =
        (current_connection_index_ == connection_count_ - 1)
        ? 0 : current_connection_index_ + 1;

      current_connection_params_ =
        connection_params_list_ [current_connection_index_];
    }
}


// Use the previous connection in the list of established connections.

void
DOVE_Supplier::use_prev_connection ()
{
  if (connection_count_ > 0)
    {
      current_connection_index_ =
        (current_connection_index_ == 0)
        ? connection_count_ - 1
        : current_connection_index_ - 1;

      current_connection_params_ =
        connection_params_list_ [current_connection_index_];
    }
}



// -------------------- Internal Demo Supplier -----------------------------

DOVE_Supplier::Internal_DOVE_Supplier::Internal_DOVE_Supplier (DOVE_Supplier *impl_ptr)
  : impl_ptr_ (impl_ptr)
{
}

// ----------------------------------------------------------------------------

int
DOVE_Supplier::get_Scheduler ()
{
  ACE_TRY_NEW_ENV
    {
      CosNaming::Name schedule_name (1);
      schedule_name.length (1);
      schedule_name[0].id =
        CORBA::string_dup (this->current_connection_params_->ss_name_);

      CORBA::Object_var objref =
          namingContext_var_->resolve (schedule_name
                                       ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      this->current_connection_params_->scheduler_var_ =
        RtecScheduler::Scheduler::_narrow(objref.in ()
                                          ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;
    }
  ACE_CATCHANY
    {
      current_connection_params_->scheduler_var_ = 0;
      ACE_ERROR_RETURN ((LM_ERROR,
                         "DOVE_Supplier::get_Scheduler: "
                         "error while resolving scheduler %s\n",
                         this->current_connection_params_->ss_name_),
                        -1);
    }
  ACE_ENDTRY;

  return 0;
}


int
DOVE_Supplier::get_EventChannel ()
{
  ACE_TRY_NEW_ENV
  {
    // Get a reference to the Event Service
    CosNaming::Name channel_name (1);
    channel_name.length (1);
    channel_name[0].id =
      CORBA::string_dup (this->current_connection_params_->es_name_);

    CORBA::Object_var eventServiceObj_var =
      this->namingContext_var_->resolve (channel_name ACE_ENV_ARG_PARAMETER);
    ACE_TRY_CHECK;

    this->current_connection_params_->eventChannel_var_ =
       RtecEventChannelAdmin::EventChannel::_narrow (eventServiceObj_var.in()
                                                   ACE_ENV_ARG_PARAMETER);
    ACE_TRY_CHECK;

    if (CORBA::is_nil (this->current_connection_params_->eventChannel_var_.in()))
      ACE_ERROR_RETURN ((LM_ERROR,
                         "The reference to the event channel is nil!"),
                         1);
  }
  ACE_CATCHANY
  {
    ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION,
                         "DOVE_Supplier::get_EventChannel");
    return -1;
  }
  ACE_ENDTRY;

  return 0;
}


int
DOVE_Supplier::connect_Supplier ()
{
  ACE_TRY_NEW_ENV
  {
    // Generate the Real-time information descriptor.
    this->current_connection_params_->rt_info_ =
      this->current_connection_params_->
        scheduler_var_->
          create (this->current_connection_params_->pod_rt_info_.entry_point
                  ACE_ENV_ARG_PARAMETER);

    ACE_TRY_CHECK;

    this->current_connection_params_->scheduler_var_->
      set (this->current_connection_params_->rt_info_,
           ACE_static_cast (RtecScheduler::Criticality_t,
                            this->current_connection_params_->pod_rt_info_.criticality),
           this->current_connection_params_->pod_rt_info_.worst_case_execution_time,
           this->current_connection_params_->pod_rt_info_.typical_execution_time,
           this->current_connection_params_->pod_rt_info_.cached_execution_time,
           this->current_connection_params_->pod_rt_info_.period,
           ACE_static_cast (RtecScheduler::Importance_t,
                            this->current_connection_params_->pod_rt_info_.importance),
           this->current_connection_params_->pod_rt_info_.quantum,
           this->current_connection_params_->pod_rt_info_.threads,
           ACE_static_cast (RtecScheduler::Info_Type_t,
                            this->current_connection_params_->pod_rt_info_.info_type)
           ACE_ENV_ARG_PARAMETER);

    ACE_TRY_CHECK;


    // Set the publications to report them to the event channel.

    CORBA::Short x = 0;
    RtecEventChannelAdmin::SupplierQOS qos;
    qos.publications.length (1);
    qos.publications[0].event.header.source = SOURCE_ID;
    qos.publications[0].event.header.type = ACE_ES_EVENT_NOTIFICATION;
    qos.publications[0].event.header.ttl = 1;
    qos.publications[0].event.header.creation_time = ORBSVCS_Time::zero ();
    qos.publications[0].event.header.ec_recv_time = ORBSVCS_Time::zero ();
    qos.publications[0].event.header.ec_send_time = ORBSVCS_Time::zero ();
    qos.publications[0].event.data.any_value <<= x;
    ACE_TRY_CHECK;
    qos.publications[0].dependency_info.number_of_calls = 1;
    qos.publications[0].dependency_info.rt_info =
      this->current_connection_params_->rt_info_;

    // = Connect as a supplier.
    this->current_connection_params_->supplierAdmin_var_ =
      this->current_connection_params_->eventChannel_var_->for_suppliers (ACE_ENV_SINGLE_ARG_PARAMETER);
    ACE_TRY_CHECK;

    this->current_connection_params_->proxyPushConsumer_var_ =
      this->current_connection_params_->supplierAdmin_var_->obtain_push_consumer (ACE_ENV_SINGLE_ARG_PARAMETER);
    ACE_TRY_CHECK;

    // In calling _this we get back an object reference and register
    // the servant with the POA.
    RtecEventComm::PushSupplier_var pushSupplier_var =
      this->internal_DOVE_Supplier_ptr_->_this (ACE_ENV_SINGLE_ARG_PARAMETER);
    ACE_TRY_CHECK;

    // Connect the supplier to the proxy consumer.
    ACE_SupplierQOS_Factory::debug (qos);
    this->current_connection_params_->
      proxyPushConsumer_var_->connect_push_supplier (pushSupplier_var.in (),
                                                     qos
                                                     ACE_ENV_ARG_PARAMETER);
    ACE_TRY_CHECK;
  }
  ACE_CATCHANY
  {
    ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION,
                         "DOVE_Supplier::connect_supplier");
    return -1;
  }
  ACE_ENDTRY;

  return 0;

}


// Access the default rt_info singleton.

ACE_Scheduler_Factory::POD_RT_Info *
DOVE_Supplier::pod_rt_info_instance ()
{
  if (DOVE_Supplier::pod_rt_info_instance_ == 0)
    {
      ACE_NEW_RETURN (DOVE_Supplier::pod_rt_info_instance_,
                      ACE_Scheduler_Factory::POD_RT_Info,
                      0);

      // Set up the default data.
      DOVE_Supplier::pod_rt_info_instance_->entry_point = "ABC";
      DOVE_Supplier::pod_rt_info_instance_->criticality =
        RtecScheduler::VERY_LOW_CRITICALITY;
      DOVE_Supplier::pod_rt_info_instance_->worst_case_execution_time =
        ORBSVCS_Time::zero ();
      DOVE_Supplier::pod_rt_info_instance_->typical_execution_time =
        ORBSVCS_Time::zero ();
      DOVE_Supplier::pod_rt_info_instance_->cached_execution_time =
        ORBSVCS_Time::zero ();
      DOVE_Supplier::pod_rt_info_instance_->period = 10000000;
      DOVE_Supplier::pod_rt_info_instance_->importance =
        RtecScheduler::VERY_LOW_IMPORTANCE;
      DOVE_Supplier::pod_rt_info_instance_->quantum = ORBSVCS_Time::zero ();
      DOVE_Supplier::pod_rt_info_instance_->threads = 1;
      DOVE_Supplier::pod_rt_info_instance_->info_type =
        RtecScheduler::OPERATION;
    }

  return DOVE_Supplier::pod_rt_info_instance_;
}