void
Notify_Push_Consumer::_connect (
  CosNotifyChannelAdmin::ConsumerAdmin_ptr consumer_admin,
  CosNotifyFilter::Filter_ptr filter)
{
  CosNotifyComm::StructuredPushConsumer_var objref =
    this->_this ();

  CosNotifyChannelAdmin::ProxySupplier_var proxysupplier =
    consumer_admin->obtain_notification_push_supplier (
    CosNotifyChannelAdmin::STRUCTURED_EVENT,
    proxy_id_);

  this->proxy_ =
    CosNotifyChannelAdmin::StructuredProxyPushSupplier::_narrow (
    proxysupplier.in ());

  CosNotification::QoSProperties properties (3);
  properties.length (3);
  properties[0].name = CORBA::string_dup (CosNotification::MaximumBatchSize);
  properties[0].value <<= (CORBA::Long) 3;
  properties[1].name = CORBA::string_dup (CosNotification::PacingInterval);
  properties[1].value <<= (TimeBase::TimeT) 1 * 1000 * 10000;

  this->proxy_->set_qos (properties);
  this->proxy_->add_filter (filter);
  this->proxy_->connect_structured_push_consumer (objref.in ());

  // give ownership to POA
  this->_remove_ref ();
}

void
TAO_Notify_ThreadPool_Consumer::connect (void)
{
  // Activate the consumer with the default_POA_
  CosNotifyComm::StructuredPushConsumer_var objref = this->_this ();

  CosNotifyChannelAdmin::ProxySupplier_var proxysupplier;

  if (this->proxy_supplier_thread_count_ != 0)
    {
      // Narrow to the extended interface.
      NotifyExt::ConsumerAdmin_var admin_ext = NotifyExt::ConsumerAdmin::_narrow (this->admin_.in ());

      NotifyExt::ThreadPoolParams tp_params = { NotifyExt::CLIENT_PROPAGATED, 0,
                                                0, static_cast<CORBA::ULong> (this->proxy_supplier_thread_count_),
                                                0, 0, 0, 0, 0 };

      CosNotification::QoSProperties qos (1);
      qos.length (1);
      qos[0].name = CORBA::string_dup (NotifyExt::ThreadPool);
      qos[0].value <<= tp_params;

      // Obtain the proxy. The QoS is applied to the POA in which the Proxy is hosted.
      proxysupplier = admin_ext->obtain_notification_push_supplier_with_qos (CosNotifyChannelAdmin::STRUCTURED_EVENT
                                                                                   , proxy_supplier_id_, qos);
    }
  else
    {
      proxysupplier = this->admin_->obtain_notification_push_supplier (CosNotifyChannelAdmin::STRUCTURED_EVENT
                                                                       , proxy_supplier_id_);
    }

  ACE_ASSERT (!CORBA::is_nil (proxysupplier.in ()));

  // narrow
  this->proxy_supplier_ =
    CosNotifyChannelAdmin::StructuredProxyPushSupplier::_narrow (proxysupplier.in ());

  ACE_ASSERT (!CORBA::is_nil (proxy_supplier_.in ()));

  this->proxy_supplier_->connect_structured_push_consumer (objref.in ());

  // Call subscription_change to inform the supplier that this consumer is available.
  CosNotification::EventTypeSeq added (1);
  CosNotification::EventTypeSeq removed;

  added.length (1);
  added[0].domain_name = CORBA::string_dup ("TEST_DOMAIN");

  /* We generate a unique Id for the consumer type so that the supplier can distinguish between the consumers.*/
  char type[BUFSIZ];
  ACE_OS::sprintf (type, "TEST_TYPE_%d", this->proxy_supplier_id_);

  added[0].type_name = CORBA::string_dup (type);

  this->proxy_supplier_->subscription_change (added, removed);

  ACE_DEBUG ((LM_DEBUG, "(%P,%t) Created Consumer %d with %d threads at the ProxySupplier\n", proxy_supplier_id_,
              this->proxy_supplier_thread_count_));
}

CosNotifyChannelAdmin::EventChannel_ptr
TAO_Notify_Lanes_Supplier_Client::create_ec (void)
{
  CosNotifyChannelAdmin::EventChannel_var ec;

  CosNotifyChannelAdmin::EventChannelFactory_var ecf = this->orb_objects_.notify_factory ();

  // Create an EventChannel
  CosNotification::QoSProperties qos;
  CosNotification::AdminProperties admin;

  // Create an event channel
  CosNotifyChannelAdmin::ChannelID id;

  ec = ecf->create_channel (qos,
                            admin,
                            id);

  // Set the Qos : 2 Lanes
  NotifyExt::ThreadPoolLanesParams tpl_params;

  tpl_params.priority_model = NotifyExt::CLIENT_PROPAGATED;
  tpl_params.server_priority = 0;
  tpl_params.stacksize = 0;
  tpl_params.lanes.length (this->consumer_count_ + 1);
  tpl_params.allow_borrowing = 0;
  tpl_params.allow_request_buffering = 0;
  tpl_params.max_buffered_requests = 0;
  tpl_params.max_request_buffer_size = 0;

  /*
   * Note that we actually create 1 extra Lane.
   * The extra Lane at priority 0 is created to match the priority 0 of the supplier thread.
   * As the ProxyConsumer is activated in an RT POA with lanes, each invocation must mach some lane.
   * Now, we typically reserve higer priorities to make requests and the lowest priority 0 for administrative calls
   * e.g. <subscription_change>. If we do not have a lane at the lowest 0 priority, then the invocation made from
   * the supplier at priority 0 will fail.
   */
  tpl_params.lanes[0].lane_priority = 0; // Priority 0
  tpl_params.lanes[0].static_threads = 1;
  tpl_params.lanes[0].dynamic_threads = 0;

  RTCORBA::Priority priority = 1; // The priority at which we send an event each.

  for (int i = 1; i <= this->consumer_count_; ++i, ++priority)
    {
      tpl_params.lanes[i].lane_priority = priority;
      tpl_params.lanes[i].static_threads = 1;
      tpl_params.lanes[i].dynamic_threads = 0;
    }

  qos.length (1);
  qos[0].name = CORBA::string_dup (NotifyExt::ThreadPoolLanes);
  qos[0].value <<= tpl_params;

  // Note that instead of <set_qos>, the <qos> can also be passed while creating the channel.
  ec->set_qos (qos);

  return ec._retn ();
}

  MyConsumer(CosNotifyChannelAdmin::EventChannel_ptr ec) :
    Miro::StructuredPushConsumer(ec)
  {
    CosNotification::QoSProperties properties;
    properties.length(2);

    // only queue one event per consumer
    properties[0].name = CORBA::string_dup(CosNotification::MaxEventsPerConsumer);
    properties[0].value <<= CORBA::Long(1);
    
    // discard older events
    properties[1].name = CORBA::string_dup(CosNotification::DiscardPolicy);
    properties[1].value <<= CORBA::Long(CosNotification::FifoOrder);

    consumerAdmin_->set_qos(properties);
  }

void
Notify_Structured_Push_Consumer::_connect (
                CosNotifyChannelAdmin::ConsumerAdmin_ptr consumer_admin)
{
#ifdef TEST_QOS_MAX_EVENTS_PER_CONSUMER
  CosNotification::QoSProperties properties (8);
  properties.length (2);
  CORBA::ULong idx = 0;
  properties[idx].name = CORBA::string_dup (CosNotification::DiscardPolicy);
  properties[idx].value <<= this->CosNotification::FifoOrder;
  idx += 1;
  properties[idx].name = CORBA::string_dup (CosNotification::MaxEventsPerConsumer);
  properties[idx].value <<= 500;
  idx += 1;
  ACE_OS::printf("Setting %d::%d QoS properties in Admin.\n", (int)idx, (int)properties.length());
  consumer_admin->set_qos (properties);
#endif // TEST_QOS_MAX_EVENTS_PER_CONSUMER

  CosNotifyComm::StructuredPushConsumer_var objref = this->_this ();

  CosNotifyChannelAdmin::ProxySupplier_var proxysupplier =
    consumer_admin->obtain_notification_push_supplier (
      CosNotifyChannelAdmin::STRUCTURED_EVENT,
      proxy_id_);

  this->proxy_ =
    CosNotifyChannelAdmin::StructuredProxyPushSupplier::_narrow (
      proxysupplier.in ());

  this->proxy_->connect_structured_push_consumer (objref.in ());

  CosNotification::EventTypeSeq added (1);
  CosNotification::EventTypeSeq removed (1);
  added.length (1);
  removed.length (1);

  added[0].domain_name = CORBA::string_dup ("*");
  added[0].type_name = CORBA::string_dup ("examples");

  removed[0].domain_name = CORBA::string_dup ("*");
  removed[0].type_name = CORBA::string_dup ("*");

  this->subscription_change (added, removed);

  // give ownership to POA
  this->_remove_ref ();
}

void
TAO_Notify_ThreadPool_Supplier::connect (void)
{
  // Activate the supplier object.
  CosNotifyComm::StructuredPushSupplier_var objref = this->_this ();

  CosNotifyChannelAdmin::ProxyConsumer_var proxyconsumer;

  if (this->proxy_consumer_thread_count_ != 0)
    {
      // Narrow to the extended interface.
      NotifyExt::SupplierAdmin_var admin_ext = NotifyExt::SupplierAdmin::_narrow (this->admin_.in ());

      NotifyExt::ThreadPoolParams tp_params = { NotifyExt::CLIENT_PROPAGATED, 0,
                                                0, static_cast<CORBA::ULong> (this->proxy_consumer_thread_count_),
                                                0, 0, 0, 0, 0 };

      CosNotification::QoSProperties qos (1);
      qos.length (1);
      qos[0].name = CORBA::string_dup (NotifyExt::ThreadPool);
      qos[0].value <<= tp_params;

      // Obtain the proxy. The QoS is applied to the POA in which the Proxy is hosted.
      proxyconsumer = admin_ext->obtain_notification_push_consumer_with_qos (CosNotifyChannelAdmin::STRUCTURED_EVENT
                                                                                   , proxy_consumer_id_, qos);
    }
  else
    {
      // Obtain the proxy.
      proxyconsumer = this->admin_->obtain_notification_push_consumer (CosNotifyChannelAdmin::STRUCTURED_EVENT
                                                                       , proxy_consumer_id_);
    }

  ACE_ASSERT (!CORBA::is_nil (proxyconsumer.in ()));

  // narrow
  this->proxy_consumer_ =
    CosNotifyChannelAdmin::StructuredProxyPushConsumer::_narrow (proxyconsumer.in ());

  ACE_ASSERT (!CORBA::is_nil (proxy_consumer_.in ()));

  // connect to the proxyconsumer.
  proxy_consumer_->connect_structured_push_supplier (objref.in ());

  ACE_DEBUG ((LM_DEBUG, "(%P,%t) Created Supplier %d with %d threads at the ProxyConsumer\n", proxy_consumer_id_,
              this->proxy_consumer_thread_count_));
}

void
Notify_Structured_Push_Consumer::_connect (
                CosNotifyChannelAdmin::ConsumerAdmin_ptr consumer_admin)
{
  CosNotifyComm::StructuredPushConsumer_var objref =
    this->_this ();

  CosNotifyChannelAdmin::ProxySupplier_var proxysupplier =
    consumer_admin->obtain_notification_push_supplier (
      CosNotifyChannelAdmin::STRUCTURED_EVENT,
      proxy_id_);

  this->proxy_ =
    CosNotifyChannelAdmin::StructuredProxyPushSupplier::_narrow (
      proxysupplier.in ());

  CosNotification::QoSProperties properties (2);
  properties.length (2);
  // The discard policy and admin properties tests already test using
  // the MaxQueueLength policy, so we'll use MEPC instead. It should
  // make no difference.
  // If the blocking timeout is less than CONSUMER_DELAY seconds, then
  // we want to ensure that exactly one event is discarded, so we set
  // the MEPC to expected_ - 1. If the supplier sends 20, then we expect 19,
  // and we set MEPC to 18. The first event will be dispatched at once, but
  // will block in the consumer. This will allow the queue in the notify svc
  // to fill up to 18. However, the blocking code will timeout before the
  // consumer finishes which will cause an event to be discarded. This will
  // allow the last event to be queued. Eventually the consumer will unblock
  // and receive the remaining events
  properties[0].name = CORBA::string_dup (CosNotification::MaxEventsPerConsumer);
  if (blocking_timeout_ < CONSUMER_DELAY * 1000 * 1000 * 10)
    properties[0].value <<= (CORBA::Long) expected_ - 1;
  else
    properties[0].value <<= (CORBA::Long) 10;


  properties[1].name = CORBA::string_dup (TAO_Notify_Extensions::BlockingPolicy);
  properties[1].value <<= this->blocking_timeout_;

  this->proxy_->set_qos (properties);
  this->proxy_->connect_structured_push_consumer (objref.in ());

  // give ownership to POA
  this->_remove_ref ();
}

void
TAO_CosNotify_Service::set_threads (CosNotification::QoSProperties &qos, int threads)
{
  NotifyExt::ThreadPoolParams tp_params =
    {NotifyExt::CLIENT_PROPAGATED, 0, 0, (unsigned)threads, 0, 0, 0, 0, 0 };

  qos.length (1);
  qos[0].name = CORBA::string_dup (NotifyExt::ThreadPool);
  qos[0].value <<= tp_params;
}

CosNotifyChannelAdmin::EventChannel_ptr
TAO_Notify_ThreadPool_Supplier_Client::create_ec (void)
{
  CosNotifyChannelAdmin::EventChannel_var ec;

  CosNotifyChannelAdmin::EventChannelFactory_var ecf = this->orb_objects_.notify_factory ();

  // Create an EventChannel
  CosNotification::QoSProperties qos;
  CosNotification::AdminProperties admin;

  // Create an event channel
  CosNotifyChannelAdmin::ChannelID id;

  ec = ecf->create_channel (qos,
                            admin,
                            id);

  // Set the Qos
  // See $TAO_ROOT/orbsvcs/orbsvcs/NotifyExt.idl
  if (this->ec_thread_count_)
    {
      NotifyExt::ThreadPoolParams tp_params = { NotifyExt::CLIENT_PROPAGATED, 0,
                                                0, static_cast<CORBA::ULong> (this->ec_thread_count_),
                                                0, 0, 0, 0, 0 };

      CosNotification::QoSProperties qos (1);
      qos.length (1);
      qos[0].name = CORBA::string_dup (NotifyExt::ThreadPool);
      qos[0].value <<= tp_params;

      // Note that instead of <set_qos>, the <qos> can also be passed while creating the channel.
      ec->set_qos (qos);
    }

  ACE_DEBUG ((LM_DEBUG, "(%P,%t) Created Event Channel with %d threads\n", this->ec_thread_count_));

  return ec._retn ();
}

void
Notify_Sequence_Push_Consumer::_connect (
  CosNotifyChannelAdmin::ConsumerAdmin_ptr consumer_admin)
{
  CosNotifyComm::SequencePushConsumer_var objref =
    this->_this ();

  CosNotifyChannelAdmin::ProxySupplier_var proxysupplier =
    consumer_admin->obtain_notification_push_supplier (
    CosNotifyChannelAdmin::SEQUENCE_EVENT,
    proxy_id_);

  this->proxy_ =
    CosNotifyChannelAdmin::SequenceProxyPushSupplier::_narrow (
    proxysupplier.in ());

  CosNotification::QoSProperties properties (3);
  properties.length (3);
  properties[0].name = CORBA::string_dup (CosNotification::MaximumBatchSize);
  properties[0].value <<= (CORBA::Long) BATCH_SIZE;
  properties[1].name = CORBA::string_dup (CosNotification::PacingInterval);
  properties[1].value <<= (TimeBase::TimeT) (1000 * 10000); // 1 secs
  if (use_ordering_)
  {
    properties[2].name = CORBA::string_dup (CosNotification::OrderPolicy);
    properties[2].value <<= this->order_policy_;
  }
  else
  {
    properties.length(2);
  }

  this->proxy_->set_qos (properties);
  this->proxy_->connect_sequence_push_consumer (objref.in ());

  // give ownership to POA
  this->_remove_ref ();
}

void
Notify_Sequence_Push_Consumer::_connect (
    CosNotifyChannelAdmin::ConsumerAdmin_ptr consumer_admin)
{
    CosNotifyComm::SequencePushConsumer_var consumer =
        this->_this ();

    CosNotifyChannelAdmin::ProxySupplier_var proxysupplier =
        consumer_admin->obtain_notification_push_supplier (
            CosNotifyChannelAdmin::SEQUENCE_EVENT,
            proxy_id_);

    this->proxy_ =
        CosNotifyChannelAdmin::SequenceProxyPushSupplier::_narrow (
            proxysupplier.in ());

    CosNotification::QoSProperties properties (4);
    properties.length (4);
    CORBA::ULong idx = 0;
    properties[idx].name = CORBA::string_dup (CosNotification::MaximumBatchSize);
    properties[idx].value <<= BATCH_SIZE;
    properties[++idx].name = CORBA::string_dup (CosNotification::PacingInterval);
    properties[idx].value <<= PACING;
    properties[++idx].name = CORBA::string_dup (CosNotification::DiscardPolicy);
    properties[idx].value <<= this->discard_policy_;
    properties[++idx].name = CORBA::string_dup (CosNotification::MaxEventsPerConsumer);
    // We set this equal to the batch size so that we conveniently always receive
    // one more batch after the first.
    properties[idx].value <<= BATCH_SIZE;

    this->proxy_->set_qos (properties);
    this->proxy_->connect_sequence_push_consumer (consumer.in ());

    // give ownership to POA
    this->_remove_ref ();
}

void
TAO_Notify_Tests_Options_Parser::execute (CosNotification::QoSProperties& qos, ACE_Arg_Shifter& arg_shifter)
{
    const ACE_TCHAR *current_arg = 0;
    NotifyExt::Priority default_priority = NotifyExt::minPriority;

    if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-ThreadPool")) == 0) // -ThreadPool [-Threads static_threads] [-Priority default_priority]
    {
        arg_shifter.consume_arg ();

        CORBA::ULong static_threads = 1u;

        if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-Threads")) == 0)
        {
            arg_shifter.consume_arg ();

            current_arg = arg_shifter.get_current ();

            static_threads = static_cast<CORBA::ULong> (ACE_OS::atoi (current_arg));

            arg_shifter.consume_arg ();
        }

        if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-Priority")) == 0)
        {
            arg_shifter.consume_arg ();
            current_arg = arg_shifter.get_current ();
            const int priority= ACE_OS::atoi (current_arg);
            if (priority < NotifyExt::minPriority)
            {
                NotifyExt::Priority default_priority = NotifyExt::minPriority;
                ACE_DEBUG ((LM_DEBUG, "-Priority %d is too small (min priority %d used)\n",
                            priority, static_cast<int> (default_priority)));
            }
            else if (NotifyExt::maxPriority < priority)
            {
                NotifyExt::Priority default_priority = NotifyExt::maxPriority;
                ACE_DEBUG ((LM_DEBUG, "-Priority %d is too large (max priority %d used)\n",
                            priority, static_cast<int> (default_priority)));
            }
            else
                default_priority = static_cast<NotifyExt::Priority> (priority);

            arg_shifter.consume_arg ();
        }

        NotifyExt::ThreadPoolParams tp_params
        = { NotifyExt::CLIENT_PROPAGATED, default_priority,
            0, static_threads, 0, default_priority, 0, 0, 0
          };

        qos.length (1);
        qos[0].name = CORBA::string_dup (NotifyExt::ThreadPool);
        qos[0].value <<= tp_params;

    } /* ThreadPool */
    else if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-Lanes")) == 0) // -Lanes lane_count -Lane prio static_thr dy_thr
    {
        arg_shifter.consume_arg ();

        current_arg = arg_shifter.get_current ();
        int lanecount = ACE_OS::atoi (current_arg);

        arg_shifter.consume_arg ();

        NotifyExt::ThreadPoolLanesParams tpl_params;

        tpl_params.priority_model = NotifyExt::CLIENT_PROPAGATED;
        tpl_params.server_priority = default_priority;
        tpl_params.stacksize = 0;
        tpl_params.lanes.length (lanecount);
        tpl_params.allow_borrowing = 0;
        tpl_params.allow_request_buffering = 0;
        tpl_params.max_buffered_requests = 0;
        tpl_params.max_request_buffer_size = 0;

        int l_index = 0;
        //parse lane values ...
        while (arg_shifter.is_anything_left ())
        {
            if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-Lane")) == 0)
            {
                arg_shifter.consume_arg ();

                // read priority
                tpl_params.lanes[l_index].lane_priority = ACE_OS::atoi (arg_shifter.get_current ());
                arg_shifter.consume_arg ();

                // static thread count
                tpl_params.lanes[l_index].static_threads = ACE_OS::atoi (arg_shifter.get_current ());
                arg_shifter.consume_arg ();

                // dynamic thread count
                tpl_params.lanes[l_index].dynamic_threads = ACE_OS::atoi (arg_shifter.get_current ());
                arg_shifter.consume_arg ();

                if (TAO_debug_level > 0)
                    ACE_DEBUG ((LM_DEBUG, "QoS Parser parsed lane: - %d, %d, %d\n",
                                tpl_params.lanes[l_index].lane_priority, tpl_params.lanes[l_index].static_threads, tpl_params.lanes[l_index].dynamic_threads));
                l_index++;
//.........这里部分代码省略.........

int
Sequence::init (int argc,
                   ACE_TCHAR* argv [])
{
  if (TAO_debug_level)
    {
      ACE_DEBUG ((LM_DEBUG,
                  "Options: event count = %d\n"
                  "supplier batch size = %d\n"
                  "consumer batch size = %d\n"
                  "pacing = %d secs\n"
                  , event_count_
                  , supplier_batch_size_
                  , consumer_batch_size_
                  , pacing_));

      ACE_DEBUG ((LM_DEBUG, "consumer delay = %d\n", consumer_delay_.sec ()));
    }

  // Initialize the base class.
  Notify_Test_Client::init (argc,
                            argv);

  // Create all participents.
  this->create_EC ();

  CosNotifyChannelAdmin::AdminID adminid;

  this->supplier_admin_ =
    this->ec_->new_for_suppliers (this->ifgop_,
                                  adminid);

  ACE_ASSERT (!CORBA::is_nil (supplier_admin_.in ()));

  this->consumer_admin_ =
    this->ec_->new_for_consumers (this->ifgop_,
                                  adminid);

  ACE_ASSERT (!CORBA::is_nil (consumer_admin_.in ()));

  ACE_NEW_RETURN (this->consumer_,
                  SequencePushConsumer (this),
                  -1);
  this->consumer_->init (root_poa_.in ());
  this->consumer_->connect (this->consumer_admin_.in ());

  CosNotification::QoSProperties properties (3);
  properties.length (3);

  properties[0].name = CORBA::string_dup (CosNotification::MaximumBatchSize);
  properties[0].value <<= (CORBA::Long) this->consumer_batch_size_;
  properties[1].name = CORBA::string_dup (CosNotification::PacingInterval);
  properties[1].value <<= (TimeBase::TimeT) (this->pacing_ * 1000 * 10000);
  properties[2].name = CORBA::string_dup (CosNotification::OrderPolicy);
  properties[2].value <<= this->order_policy_;

  this->consumer_->get_proxy_supplier ()->set_qos (properties);

  ACE_NEW_RETURN (this->supplier_,
                  SequencePushSupplier (this),
                  -1);
  this->supplier_->init (root_poa_.in ());

  this->supplier_->connect (this->supplier_admin_.in ());

  consumer_start( 0 );

  return 0;
}

    //------------------------------------------------------
    CosNotification::QoSProperties
    CDBProperties::getCDBQoSProps(const std::string& channelName)
    {
	CosNotification::QoSProperties retVal;
	retVal.length(0);
	
	//sanity check
	if (cdbChannelConfigExists(channelName)==false)
	    {
	    return retVal;
	    }
	
	//CDB
	//complete name of the channel within the CDB
	std::string cdbChannelName = "MACI/Channels/" + channelName;
	CDB::DAL_var cdbRef = getCDB();
	CDB::DAO_var tempDAO = cdbRef->get_DAO_Servant(cdbChannelName.c_str());

	//temporary pointer points to the name of the CosNotification
	//property
	const char *name_p;
	//temporary counter
	unsigned int i = 0U;
	
	//commented out because TAO does not support these props
	/*
	//EventReliability///////////////////////////////////////////////
	{
	name_p = CosNotification::EventReliability;
	//allocate one extra element
	i++;
	retVal.length(i);
	retVal[i-1].name = CORBA::string_dup(name_p);
	std::string tStringER = tempDAO->get_string(name_p);
	if(tStringER=="BestEffort")
	    {
	    retVal[i-1].value <<= CosNotification::BestEffort; 
	    }
	else
	    {
	    retVal[i-1].value <<= CosNotification::Persistent;
	    }
	}
	//ConnectionReliability///////////////////////////////////////////////
	{
	name_p = CosNotification::ConnectionReliability;
	//allocate one extra element
	i++;
	retVal.length(i);
	retVal[i-1].name = CORBA::string_dup(name_p);
	std::string tStringCR = tempDAO->get_string(name_p);
	if(tStringCR=="BestEffort")
	    {
	    retVal[i-1].value <<= CosNotification::BestEffort; 
	    }
	else
	    {
	    retVal[i-1].value <<= CosNotification::Persistent;
	    }
	}
	*/
	//Priority/////////////////////////////////////////////////////
	{
	name_p = CosNotification::Priority;
	//allocate one extra element
	i++;
	retVal.length(i);
	retVal[i-1].name = CORBA::string_dup(name_p);
	retVal[i-1].value <<= static_cast<CORBA::Short>(tempDAO->get_long(name_p));
	}
	//Timeout//////////////////////////////////////////////////////
	{
	name_p = CosNotification::Timeout;
	//allocate one extra element
	i++;
	retVal.length(i);
	retVal[i-1].name = CORBA::string_dup(name_p);
	retVal[i-1].value <<= static_cast<TimeBase::TimeT>(tempDAO->get_long(name_p));
	}
	//OrderPolicy///////////////////////////////////////////////
	{
	name_p = CosNotification::OrderPolicy;
	//allocate one extra element
	i++;
	retVal.length(i);
	retVal[i-1].name = CORBA::string_dup(name_p);
	std::string tStringOP = tempDAO->get_string(name_p);
	if(tStringOP=="AnyOrder")
	    {
	    retVal[i-1].value <<= CosNotification::AnyOrder; 
	    }
	else if(tStringOP=="FifoOrder")
	    {
	    retVal[i-1].value <<= CosNotification::FifoOrder; 
	    }
	else if(tStringOP=="PriorityOrder")
	    {
	    retVal[i-1].value <<= CosNotification::PriorityOrder; 
	    }
//.........这里部分代码省略.........

int ACE_TMAIN (int argc, ACE_TCHAR *argv[])
{
    int status = 0;
    ACE_Auto_Ptr< sig_i > sig_impl;
    try
    {
        Supplier_Client client;
        status = client.init (argc, argv);

        if (status == 0)
        {
            CosNotifyChannelAdmin::EventChannel_var ec =
                client.create_event_channel ("MyEventChannel", 0);

            if (use_deadline_ordering)
            {
                CosNotification::QoSProperties qos (1);
                qos.length (1);
                qos[0].name = CORBA::string_dup (CosNotification::OrderPolicy);
                qos[0].value <<= (CORBA::Short)CosNotification::DeadlineOrder;
                ec->set_qos (qos);
            }

            sig_impl.reset( new sig_i( client.orb() ) );
            sig_var sig = sig_impl->_this ();

            // If the ior_output_file exists, output the ior to it
            if (ior_output_file != 0)
            {
                CORBA::String_var ior =
                    client.orb ()->object_to_string (sig.in ());

                FILE *output_file= ACE_OS::fopen (ior_output_file, "w");
                ACE_ASSERT (output_file != 0);
                ACE_OS::fprintf (output_file, "%s", ior.in ());
                ACE_OS::fclose (output_file);
            }

            CosNotifyChannelAdmin::SupplierAdmin_var admin =
                create_supplieradmin (ec.in ());
            ACE_ASSERT(!CORBA::is_nil (admin.in ()));

            create_suppliers (admin.in (), client.root_poa ());

            sig_impl->wait_for_startup();

            ACE_DEBUG((LM_DEBUG, "1 supplier sending %d events...\n", num_events));
            for (int i = 0; i < num_events; ++i)
            {
                ACE_DEBUG((LM_DEBUG, "+"));
                SendEvent (i + 1);
            }
            ACE_DEBUG((LM_DEBUG, "\nSupplier sent %d events.\n", num_events));

            sig_impl->wait_for_completion();

            ACE_OS::unlink (ior_output_file);

            supplier_1->disconnect();

            ec->destroy();
        }
    }
    catch (const CORBA::Exception& e)
    {
        e._tao_print_exception ("Error: ");
        status = 1;
    }

    return status;
}

int
ACE_TMAIN (int argc, ACE_TCHAR *argv[])
{
  try
    {
      CORBA::ORB_var orb = CORBA::ORB_init(argc, argv);

      CORBA::Object_var naming_obj =
        orb->resolve_initial_references ("NameService");

      CosNaming::NamingContext_var naming_context =
        CosNaming::NamingContext::_narrow(naming_obj.in());

      CosNaming::Name name;
      name.length (1);
      name[0].id = CORBA::string_dup("MyEventChannel");
      CORBA::Object_var ecObj = naming_context->resolve(name);

      CosNotifyChannelAdmin::EventChannel_var ec =
        CosNotifyChannelAdmin::EventChannel::_narrow(ecObj.in());

      CosNotifyChannelAdmin::AdminID adminid;
      CosNotifyChannelAdmin::InterFilterGroupOperator ifgop =
        CosNotifyChannelAdmin::AND_OP;

      CosNotifyChannelAdmin::ConsumerAdmin_var consumer_admin =
        ec->new_for_consumers(ifgop, adminid);

      CORBA::Object_var poa_object =
        orb->resolve_initial_references("RootPOA");

      PortableServer::POA_var poa =
        PortableServer::POA::_narrow (poa_object.in());

      PortableServer::Servant_var<StructuredEventConsumer_i> servant =
        new StructuredEventConsumer_i(orb.in());

      PortableServer::ObjectId_var objectId = poa->activate_object (servant.in());

      CORBA::Object_var consumer_obj = poa->id_to_reference (objectId.in ());

      CosNotifyComm::StructuredPushConsumer_var consumer =
        CosNotifyComm::StructuredPushConsumer::_narrow (consumer_obj.in ());

      CosNotifyChannelAdmin::ProxyID consumeradmin_proxy_id;

      CosNotifyChannelAdmin::ProxySupplier_var proxy_supplier =
        consumer_admin->obtain_notification_push_supplier(
          CosNotifyChannelAdmin::STRUCTURED_EVENT,
          consumeradmin_proxy_id);

      CosNotifyChannelAdmin::StructuredProxyPushSupplier_var supplier_proxy;
      supplier_proxy =
      CosNotifyChannelAdmin::StructuredProxyPushSupplier::_narrow(proxy_supplier.in());

      CosNotification::QoSProperties properties (1);

      properties.length (1);
      properties[0].name = CORBA::string_dup (CosNotification::OrderPolicy);
      properties[0].value <<= CosNotification::FifoOrder;

      supplier_proxy->set_qos (properties);
      supplier_proxy->connect_structured_push_consumer(consumer.in());

      CosNotification::EventTypeSeq added (1);
      CosNotification::EventTypeSeq removed (1);
      added.length (1);
      removed.length (1);

      added[0].domain_name = CORBA::string_dup ("OCI_TAO");
      added[0].type_name = CORBA::string_dup ("examples");

      removed[0].domain_name = CORBA::string_dup ("*");
      removed[0].type_name = CORBA::string_dup ("*");

      supplier_proxy->subscription_change(added, removed);

      PortableServer::POAManager_var poa_manager = poa->the_POAManager();

      poa_manager->activate();

      orb->run();
    }
  catch(const CORBA::Exception& ex)
    {
      std::cerr << "Caught exception: " << ex << std::endl;
        return 1;
    }

  return 0;
}

//.........这里部分代码省略.........

      // Create a thread-pool policy.
      CORBA::Policy_var lanes_policy =
        rt_orb->create_threadpool_policy (threadpool_id);

      CORBA::PolicyList poa_policy_list(2);
      poa_policy_list.length (2);
      poa_policy_list[0] = priority_model_policy;
      poa_policy_list[1] = lanes_policy;

      PortableServer::POAManager_var poa_manager = poa->the_POAManager ();

      PortableServer::POA_var rt_poa = poa->create_POA ("RT POA",
              poa_manager.in (),
              poa_policy_list);

      PortableServer::Servant_var<StructuredEventConsumer_i> servant =
        new StructuredEventConsumer_i(orb.in());

      PortableServer::ObjectId_var objectId =
        rt_poa->activate_object (servant.in());

      CORBA::Object_var consumer_obj =
        rt_poa->id_to_reference (objectId.in ());

      CosNotifyComm::StructuredPushConsumer_var consumer =
        CosNotifyComm::StructuredPushConsumer::_narrow (consumer_obj.in ());

      NotifyExt::ThreadPoolLanesParams tpl_params;

      tpl_params.priority_model = NotifyExt::CLIENT_PROPAGATED;
      tpl_params.server_priority = DEFAULT_PRIORITY;
      tpl_params.stacksize = 0;
      tpl_params.allow_borrowing = 0;
      tpl_params.allow_request_buffering = 0;
      tpl_params.max_buffered_requests = 0;
      tpl_params.max_request_buffer_size = 0;
      tpl_params.lanes.length (2);
      tpl_params.lanes[0].lane_priority   = LOW_PRIORITY;
      tpl_params.lanes[0].static_threads  = 2;
      tpl_params.lanes[0].dynamic_threads = 0;
      tpl_params.lanes[1].lane_priority   = HIGH_PRIORITY;
      tpl_params.lanes[1].static_threads  = 2;
      tpl_params.lanes[1].dynamic_threads = 0;
      CosNotification::QoSProperties qos;
      qos.length(1);
      qos[0].name = CORBA::string_dup (NotifyExt::ThreadPoolLanes);
      qos[0].value <<= tpl_params;

      consumer_admin->set_qos(qos);
      CORBA::Object_var current_obj =
        orb->resolve_initial_references ("RTCurrent");

      RTCORBA::Current_var current =
        RTCORBA::Current::_narrow (current_obj.in ());
      current->the_priority(HIGH_PRIORITY);

      CosNotifyChannelAdmin::ProxyID consumeradmin_proxy_id;

      CosNotifyChannelAdmin::ProxySupplier_var proxy_supplier =
        consumer_admin->obtain_notification_push_supplier(
          CosNotifyChannelAdmin::STRUCTURED_EVENT,
          consumeradmin_proxy_id);

      CosNotifyChannelAdmin::StructuredProxyPushSupplier_var supplier_proxy;
      supplier_proxy = CosNotifyChannelAdmin::StructuredProxyPushSupplier::
        _narrow(proxy_supplier.in());

      supplier_proxy->connect_structured_push_consumer(consumer.in());

      CosNotification::EventTypeSeq added (1);
      CosNotification::EventTypeSeq removed (1);
      added.length (1);
      removed.length (1);

      added[0].domain_name = CORBA::string_dup ("OCI_TAO");
      added[0].type_name = CORBA::string_dup ("examples");

      removed[0].domain_name = CORBA::string_dup ("*");
      removed[0].type_name = CORBA::string_dup ("*");

      supplier_proxy->subscription_change(added, removed);

      poa_manager->activate();

      // Write a file to let the run_test.pl script know we are ready.
      std::ofstream iorFile( ACE_TEXT_ALWAYS_CHAR(output_file) );
      iorFile << "Ready" << std::endl;
      iorFile.close();

      orb->run();
    }
  catch(const CORBA::Exception& ex)
    {
      std::cerr << "Caught exception: " << ex << std::endl;
        return 1;
    }

  return 0;
}

int ACE_TMAIN (int argc, ACE_TCHAR *argv[])
{
  ACE_Auto_Ptr< sig_i > sig_impl;
  try
  {
    Supplier_Client client;
    int status = client.init (argc, argv);
    ACE_UNUSED_ARG(status);
    ACE_ASSERT(status == 0);

    CosNotifyChannelAdmin::EventChannel_var ec =
      client.create_event_channel ("MyEventChannel", 0);

    CosNotification::QoSProperties qos (1);
    qos.length (1);
    qos[0].name = CORBA::string_dup (CosNotification::OrderPolicy);
    qos[0].value <<= order_policy;
    ec->set_qos (qos);

    CORBA::ORB_ptr orb = client.orb ();

    sig_impl.reset( new sig_i( orb ) );
    sig_var sig = sig_impl->_this ();

    CORBA::String_var ior =
      orb->object_to_string (sig.in ());

    if (ior_output_file != 0)
    {
      FILE *output_file= ACE_OS::fopen (ior_output_file, "w");
      if (output_file == 0)
        ACE_ERROR_RETURN ((LM_ERROR,
        "Cannot open output file %s for "
        "writing IOR: %C",
        ior_output_file,
        ior.in ()),
        1);
      ACE_OS::fprintf (output_file, "%s", ior.in ());
      ACE_OS::fclose (output_file);
    }

    CosNotifyChannelAdmin::SupplierAdmin_var admin =
      create_supplieradmin (ec.in ());
    ACE_ASSERT(!CORBA::is_nil (admin.in ()));
    create_suppliers (admin.in (), client.root_poa ());

    sig_impl->wait_for_startup();

    ACE_DEBUG((LM_DEBUG, "1 supplier sending %d events...\n", num_events));
    for (int i = 0; i < num_events / BATCH_SIZE; ++i)
    {
      ACE_DEBUG((LM_DEBUG, "+"));
      SendBatch (i);
    }
    ACE_DEBUG((LM_DEBUG, "\nSupplier sent %d events.\n", num_events));

    sig_impl->wait_for_completion();

    ACE_OS::unlink (ior_output_file);

    supplier_1->disconnect();

    ec->destroy();

    return 0;
  }
  catch (const CORBA::Exception& e)
  {
    e._tao_print_exception ("Error: ");
  }

  return 1;
}

void
Supplier_Main::init_event_channel (void)
{
  bool ok = false;
  if (this->reconnecting_)
  {
    try
    {
      this->ec_ = this->ecf_->get_event_channel (
            this->ec_id_);
      ok = ! CORBA::is_nil (this->ec_.in ());
      if (ok && this->verbose_)
      {
        ACE_DEBUG ((LM_DEBUG,
          ACE_TEXT ("(%P|%t) Supplier: Reconnect to event channel %d\n"),
          static_cast<int>(this->ec_id_)
          ));
      }
    }
    catch (...)
    {
    }
  }

  // if we don't have a channel yet, and a channel id file was specified
  // try to read from it
  if (!ok && this->channel_file_.length () > 0)
  {
    FILE * chf = ACE_OS::fopen (this->channel_file_.c_str (), "r");
    if (chf != 0)
    {
      char buffer[100];
      ACE_OS::fgets (buffer, sizeof(buffer), chf);
      ACE_OS::fclose (chf);
      this->ec_id_ = ACE_OS::atoi (buffer);

      try
      {
        this->ec_ = this->ecf_->get_event_channel (
              this->ec_id_);
        ok = ! CORBA::is_nil (this->ec_.in ());
        if (ok)
        {
          if (this->verbose_)
          {
            ACE_DEBUG ((LM_DEBUG,
              ACE_TEXT ("(%P|%t) Supplier: Connect to Existing event channel %d\n"),
              static_cast<int>(this->ec_id_)