int
TAO_AV_Callback::handle_stop (void)
{
  if (TAO_debug_level > 0) ORBSVCS_DEBUG ((LM_DEBUG,"TAO_AV_Callback::handle_stop\n"));
  return -1;
}

int
TAO_AV_Core::init_forward_flows (TAO_Base_StreamEndPoint *endpoint,
                                 TAO_AV_FlowSpecSet &flow_spec_set,
                                 TAO_AV_Core::EndPoint direction,
                                 AVStreams::flowSpec &flow_spec)
{
  if (TAO_debug_level > 0)
    ORBSVCS_DEBUG ((LM_DEBUG,
                "TAO_AV_Core::init_forward_flows\n"));

  TAO_AV_FlowSpecSet address_flow_set;
  TAO_AV_FlowSpecSet flow_set;
  TAO_AV_FlowSpecSetItor end = flow_spec_set.end ();
  for (TAO_AV_FlowSpecSetItor start = flow_spec_set.begin ();
       start != end; ++start)
    {
      TAO_FlowSpec_Entry *entry = (*start);
      switch (direction)
        {
        case TAO_AV_Core::TAO_AV_ENDPOINT_B:
          {
            switch (entry->direction ())
              {
              case TAO_FlowSpec_Entry::TAO_AV_DIR_IN:
                {
                  entry->role (TAO_FlowSpec_Entry::TAO_AV_CONSUMER);
                  break;
                }
              case TAO_FlowSpec_Entry::TAO_AV_DIR_OUT:
                {
                  entry->role (TAO_FlowSpec_Entry::TAO_AV_PRODUCER);
                  break;
                }
              }
            break;
          }
        case TAO_AV_Core::TAO_AV_ENDPOINT_A:
          {
            switch (entry->direction ())
              {
              case TAO_FlowSpec_Entry::TAO_AV_DIR_IN:
                entry->role (TAO_FlowSpec_Entry::TAO_AV_PRODUCER);
                break;
              case TAO_FlowSpec_Entry::TAO_AV_DIR_OUT:
                entry->role (TAO_FlowSpec_Entry::TAO_AV_CONSUMER);
                break;
              }
            break;
          }
        default:
          break;
        }
      ACE_Addr *address = entry->address ();
      if (address != 0)
        {
          if (TAO_debug_level > 0)
            ORBSVCS_DEBUG ((LM_DEBUG,
                        "address given for flow %s\n",
                        entry->flowname ()));

          address_flow_set.insert (entry);
        }
      else
        flow_set.insert (entry);
    } //End of For Loop


  int result = -1;
  switch (direction)
    {
    case TAO_AV_Core::TAO_AV_ENDPOINT_A:
      if (address_flow_set.size () > 0)
        {
          result = this->acceptor_registry_->open (endpoint,
                                                   this,
                                                   address_flow_set);
          if (result < 0)
            ORBSVCS_ERROR_RETURN ((LM_ERROR,
                               "TAO_AV_Core::init_forward_flows::acceptor_registry::open failed\n"),
                              -1);
          TAO_AV_FlowSpecSetItor end = address_flow_set.end ();
          for (TAO_AV_FlowSpecSetItor start = address_flow_set.begin ();
               start != end; ++start)
            {
              TAO_FlowSpec_Entry *entry = (*start);
              switch (entry->direction ())
                {
                case TAO_FlowSpec_Entry::TAO_AV_DIR_IN:
                  {
                    if (entry->handler () != 0)
                      {
                        //Yamuna:PLEASE CHECK THIS LATER
#if defined ACE_HAS_RAPI || defined (ACE_HAS_WINSOCK2_GQOS)
                        // For IN flows on the A side we should remove the handlers from the reactor.
                        ACE_Event_Handler *event_handler = entry->handler ()->event_handler ();

                        if (event_handler->reactor () != 0)
                          {
                            result = event_handler->reactor ()->remove_handler (event_handler,
                                                                                ACE_Event_Handler::READ_MASK);
//.........这里部分代码省略.........

void
AsyncListManager::final_state (void)
{
  if (ImR_Locator_i::debug() > 4)
    {
      ORBSVCS_DEBUG ((LM_DEBUG,
                      ACE_TEXT ("(%P|%t) AsyncListManager(%@)::final_state, ")
                      ACE_TEXT ("waiters count = %d, has pinger? %d\n"),
                      this, this->waiters_, (this->pinger_ != 0)));
    }

  if (this->pinger_ != 0 && this->waiters_ > 0)
    {
      return;
    }

  bool excepted = false;
  CORBA::ULong len = this->server_list_.length ();
  ImplementationRepository::ServerInformationList alt_list (this->how_many_);
  ImplementationRepository::ServerInformationList *sil = &this->server_list_;
  if (this->first_ > 0 || this->how_many_ < len)
    {
      alt_list.length (this->how_many_);
      for (CORBA::ULong i = 0; i < this->how_many_; i++)
        {
          alt_list[i] = this->server_list_[i + this->first_];
        }
      sil = &alt_list;
    }

  if (!CORBA::is_nil (this->primary_.in ()))
    {
      ImplementationRepository::ServerInformationIterator_var server_iterator;
      if (sil !=  &this->server_list_)
        {
          excepted = !this->make_iterator (server_iterator.out(), this->how_many_);
        }
      else
        {
          server_iterator =
            ImplementationRepository::ServerInformationIterator::_nil ();
        }

      if (!excepted)
        {
          try
            {
              this->primary_->list (*sil, server_iterator.in ());
            }
          catch (CORBA::Exception &ex)
            {
              ex._tao_print_exception ("AsyncListManager:final state sending list\n");
              ImplementationRepository::AMH_AdministrationExceptionHolder h (ex._tao_duplicate());
              this->primary_->list_excep (&h);
            }
        }
      this->primary_ =
        ImplementationRepository::AMH_AdministrationResponseHandler::_nil ();
    }
  else if (!CORBA::is_nil (this->secondary_.in()))
    {
      CORBA::Boolean done = this->first_ + sil->length() == len;
      try
        {
          this->secondary_->next_n (done, *sil);
        }
      catch (CORBA::Exception &ex)
        {
          ex._tao_print_exception ("AsyncListManager:final state sending secondary list\n");
          ImplementationRepository::AMH_ServerInformationIteratorExceptionHolder h (ex._tao_duplicate());
          this->secondary_->next_n_excep (&h);
        }
      this->secondary_ =
        ImplementationRepository::AMH_ServerInformationIteratorResponseHandler::_nil ();
    }
}

CORBA::ParDescriptionSeq *
TAO_OperationDef_i::params_i (void)
{
  CORBA::ULong i = 0;
  ACE_Configuration_Section_Key params_key;
  int status =
    this->repo_->config ()->open_section (this->section_key_,
                                          "params",
                                          0,
                                          params_key);

  ACE_Unbounded_Queue<ACE_Configuration_Section_Key> key_queue;

  // This section may not have been created.
  if (status == 0)
    {
      u_int count = 0;
      this->repo_->config ()->get_integer_value (params_key,
                                                 "count",
                                                 count);

      for (i = 0; i < count; ++i)
        {
          ACE_Configuration_Section_Key param_key;
          char *stringified = TAO_IFR_Service_Utils::int_to_string (i);
          status =
            this->repo_->config ()->open_section (params_key,
                                                  stringified,
                                                  0,
                                                  param_key);

          if (status == 0)
            {
              key_queue.enqueue_tail (param_key);
            }
        }
    }

  CORBA::ULong size = static_cast<CORBA::ULong> (key_queue.size ());

  CORBA::ParDescriptionSeq *pd_seq = 0;
  ACE_NEW_THROW_EX (pd_seq,
                    CORBA::ParDescriptionSeq (size),
                    CORBA::NO_MEMORY ());

  pd_seq->length (size);

  if (size == 0)
    {
      return pd_seq;
    }

  CORBA::ParDescriptionSeq_var retval = pd_seq;

  for (i = 0; i < size; ++i)
    {
      ACE_Configuration_Section_Key key;
      key_queue.dequeue_head (key);

      ACE_TString name;
      this->repo_->config ()->get_string_value (key,
                                                "name",
                                                name);

      retval[i].name = name.c_str ();

      u_int mode = 0;
      this->repo_->config ()->get_integer_value (key,
                                                 "mode",
                                                 mode);

      retval[i].mode = static_cast<CORBA::ParameterMode> (mode);

      ACE_TString type_path;
      this->repo_->config ()->get_string_value (key,
                                                "type_path",
                                                type_path);

      TAO_IDLType_i *impl =
        TAO_IFR_Service_Utils::path_to_idltype (type_path,
                                                this->repo_);

      // We have an undefined interface
      if (impl == 0)
        {
          ORBSVCS_DEBUG ((LM_DEBUG,
                      ACE_TEXT ("Encountered undefined parameter type "
                                "for variable: %s\n"),
                      name.c_str ()));

          throw CORBA::INTF_REPOS ();
        }

      retval[i].type = impl->type_i ();

      CORBA::Object_var obj =
        TAO_IFR_Service_Utils::path_to_ir_object (type_path,
                                                  this->repo_);

      retval[i].type_def = CORBA::IDLType::_narrow (obj.in ());
//.........这里部分代码省略.........

int
Task_Entry::conjunctive_merge (Dependency_Type dt,
                               ACE_Unbounded_Set <Dispatch_Entry *> &dispatch_entries,
                               ACE_CString &unresolved_locals,
                               ACE_CString &unresolved_remotes)
{
  int result = 0;
  char string_buffer [BUFSIZ];

  // Iterate over the dependencies, and determine the total frame
  // size.

  u_long frame_size = 1;

  ACE_Unbounded_Set_Iterator <Task_Entry_Link *> dep_iter (callers_);

  for (dep_iter.first ();
       dep_iter.done () == 0;
       dep_iter.advance ())
    {
      Task_Entry_Link **link;

      if (dep_iter.next (link) == 0
          || link == 0
          || *link == 0)
        return -1;

      // The link matches the dependency type given.
      if ((*link)->dependency_type () == dt)
        {
          // Check for and warn about unresolved remote dependencies
          // in the ONE_WAY call graph.
          if ((*link)->dependency_type () == RtecBase::ONE_WAY_CALL
              && (*link)->caller ().has_unresolved_remote_dependencies ()
              && ! this->has_unresolved_remote_dependencies ())
            {
              // Propagate the unresolved remote dependency flag, and
              // issue a debug scheduler warning.
              this->has_unresolved_remote_dependencies (1);
              ORBSVCS_DEBUG ((LM_DEBUG,
                          "Warning: an operation identified by "
                          "\"%s\" has unresolved remote dependencies.\n",
                          (const char*) this->rt_info ()->entry_point));

              // Record entry point in list of unresolved remote
              // dependencies
              ACE_OS::sprintf (string_buffer,
                               "// %s\n",
                               (const char*) this->rt_info ()->entry_point);
              unresolved_remotes +=
                ACE_CString (string_buffer);
            }

          // Check for and warn about unresolved local dependencies in
          // the ONE_WAY call graph.
          if ((*link)->dependency_type () == RtecBase::ONE_WAY_CALL
              && (*link)->caller ().has_unresolved_local_dependencies ()
              && ! this->has_unresolved_local_dependencies ())
            {
              // Propagate the unresolved local dependency flag, and
              // issue a debug scheduler warning.
              this->has_unresolved_local_dependencies (1);
              ORBSVCS_DEBUG ((LM_DEBUG,
                          "Warning: an operation identified by "
                          "\"%s\" has unresolved local dependencies.\n",
                          (const char*) this->rt_info ()->entry_point));

              // Record entry point in list of unresolved local dependencies
              ACE_OS::sprintf (string_buffer,
                               "// %s\n",
                               (const char*) this->rt_info ()->entry_point);
              unresolved_locals +=
                ACE_CString (string_buffer);
            }

          frame_size = ACE::minimum_frame_size (frame_size,
                                                (*link)->caller ().effective_period_);
        }
    }

  // Reframe dispatches in the set to the new frame size (expands the
  // set's effective period to be the new enclosing frame).
  if (reframe (dispatch_entries,
               *this, dispatches_,
               effective_period_,
               frame_size) < 0)
    return -1;

  // A container and iterator for virtual dispatch sets over which the
  // conjunction will operate
  ACE_Ordered_MultiSet <Dispatch_Proxy_Iterator *> conj_set;
  ACE_Ordered_MultiSet_Iterator <Dispatch_Proxy_Iterator *> conj_set_iter (conj_set);

  // Iterate over the dependencies, and for each of the given call
  // type, create a Dispatch_Proxy_Iterator for the caller's dispatch
  // set, using the caller's period, the total frame size, and the
  // number of calls: if any of the sets is empty, just return 0;
  for (dep_iter.first ();
       dep_iter.done () == 0;
       dep_iter.advance ())
//.........这里部分代码省略.........

void
TAO_ECG_CDR_Message_Sender::send_fragment (const ACE_INET_Addr &addr,
                                           CORBA::ULong request_id,
                                           CORBA::ULong request_size,
                                           CORBA::ULong fragment_size,
                                           CORBA::ULong fragment_offset,
                                           CORBA::ULong fragment_id,
                                           CORBA::ULong fragment_count,
                                           iovec iov[],
                                           int iovcnt)
{
  CORBA::ULong header[TAO_ECG_CDR_Message_Sender::ECG_HEADER_SIZE
                     / sizeof(CORBA::ULong)
                     + ACE_CDR::MAX_ALIGNMENT];
  char* buf = reinterpret_cast<char*> (header);
  TAO_OutputCDR cdr (buf, sizeof(header));
  cdr.write_boolean (TAO_ENCAP_BYTE_ORDER);
  // Insert some known values in the padding bytes, so we can smoke
  // test the message on the receiving end.
  cdr.write_octet ('A'); cdr.write_octet ('B'); cdr.write_octet ('C');
  cdr.write_ulong (request_id);
  cdr.write_ulong (request_size);
  cdr.write_ulong (fragment_size);
  cdr.write_ulong (fragment_offset);
  cdr.write_ulong (fragment_id);
  cdr.write_ulong (fragment_count);
  CORBA::Octet padding[4];


  // MRH
  if (checksum_)
  {
    // Compute CRC
     iov[0].iov_base = cdr.begin ()->rd_ptr ();
     iov[0].iov_len  = cdr.begin ()->length ();
     unsigned int crc = 0;
     unsigned char *crc_parts = (unsigned char *)(&crc);
     if (iovcnt > 1)
       {
         crc = ACE::crc32 (iov, iovcnt);
         crc = ACE_HTONL (crc);
       }
     for (int cnt=0; cnt<4; ++cnt)
       {
         padding[cnt] = crc_parts[cnt];
       }
   }
   else
   {
     for (int cnt=0; cnt<4; ++cnt)
     {
       padding[cnt] = 0;
     }
   }
   //End MRH
  cdr.write_octet_array (padding, 4);

  iov[0].iov_base = cdr.begin ()->rd_ptr ();
  iov[0].iov_len  = cdr.begin ()->length ();

  ssize_t n = this->dgram ().send (iov,
                                   iovcnt,
                                   addr);
  size_t expected_n = 0;
  for (int i = 0; i < iovcnt; ++i)
    expected_n += iov[i].iov_len;
  if (n > 0 && size_t(n) != expected_n)
    {
      ORBSVCS_ERROR ((LM_ERROR, ("Sent only %d out of %d bytes "
                              "for mcast fragment.\n"),
                  n,
                  expected_n));
    }

  if (n == -1)
    {
      if (errno == EWOULDBLOCK)
        {
          ORBSVCS_ERROR ((LM_ERROR, "Send of mcast fragment failed (%m).\n"));
          // @@ TODO Use a Event Channel specific exception
          throw CORBA::COMM_FAILURE ();
        }
      else
        {
          ORBSVCS_DEBUG ((LM_WARNING, "Send of mcast fragment blocked (%m).\n"));
        }
    }
  else if (n == 0)
    {
      ORBSVCS_DEBUG ((LM_WARNING, "EOF on send of mcast fragment (%m).\n"));
    }
}

int
TAO::PG_Object_Group::set_primary_member (
    TAO_IOP::TAO_IOR_Property * prop,
    const PortableGroup::Location & the_location)
{
  ACE_GUARD_RETURN (TAO_SYNCH_MUTEX,
                    guard,
                    this->internals_,
                    0);
  int result = 1;
  MemberInfo * info = 0;
  if (this->members_.find (the_location, info) == 0)
    {
      int cleared = 0;
      this->primary_location_ = the_location;
      for (MemberMap_Iterator it = this->members_.begin ();
           !cleared && it != this->members_.end ();
           ++it)
        {
          cleared = (*it).int_id_->is_primary_;
          (*it).int_id_->is_primary_ = 0;
        }
      info->is_primary_ = 1;

      int set_ok =
        this->manipulator_.set_primary (prop,
                                        this->reference_.in (),
                                        info->member_.in ());
      if (!set_ok)
        {
          if (TAO_debug_level > 3)
            {
              ORBSVCS_ERROR ((LM_ERROR,
                          ACE_TEXT ("%T %n (%P|%t) - ")
                          ACE_TEXT ("Can't set primary in IOGR .\n")
                          ));
            }
//@@: ACE_THROW (FT::PrimaryNotSet());
          result = 0;
        }

      if (result && this->increment_version ())
        {
          this->distribute_iogr ();
        }
      else
        {
          if (TAO_debug_level > 3)
            {
              ORBSVCS_DEBUG ((LM_DEBUG,
                          ACE_TEXT("TAO-PG (%P|%t) - set_primary_location ")
                          ACE_TEXT("throwing PrimaryNotSet because increment")
                          ACE_TEXT("version failed.\n")
                          ));
            }
//@@: ACE_THROW (FT::PrimaryNotSet());
          result = 0;
        }
    }
  else
    {
      if (TAO_debug_level > 3)
        {
          ORBSVCS_DEBUG ((LM_DEBUG,
                      ACE_TEXT ("TAO-PG (%P|%t) - set_primary_location ")
                      ACE_TEXT ("throwing MemberNotFound.\n")));
        }
      throw PortableGroup::MemberNotFound();
    }

  return result;
}

//.........这里部分代码省略.........
            {
              // The context factory is now owned by the activator
              // so we should release it
              contextFactory.release ();
              // If using a servant activator, the activator now owns the
              // factory, so we should release it
              persFactory.release ();
            }

      }
      else if (persistence_location != 0)
        //
        // Initialize Persistent Naming Service.
        //
        {

          // Create Naming Context Implementation Factory to be used for the creation of
          // naming contexts by the TAO_Persistent_Context_Index
          TAO_Persistent_Naming_Context_Factory *naming_context_factory =
            this->persistent_naming_context_factory ();
          // Make sure we got a factory.
          if (naming_context_factory == 0) return -1;

          // Allocate and initialize Persistent Context Index.
          ACE_NEW_RETURN (this->context_index_,
                          TAO_Persistent_Context_Index (orb, poa, naming_context_factory),
                          -1);

          if (this->context_index_->open (persistence_location,
                                          base_addr) == -1
              || this->context_index_->init (context_size) == -1)
            {
              if (TAO_debug_level >0)
                ORBSVCS_DEBUG ((LM_DEBUG,
                            "TAO_Naming_Server: context_index initialization failed\n"));
              return -1;
            }

          // Set the root Naming Context reference.
          this->naming_context_ =
            this->context_index_->root_context ();
        }
      else
#endif /* CORBA_E_MICRO */
        {
          //
          // Initialize Transient Naming Service.
          //
          this->naming_context_ =
            TAO_Transient_Naming_Context::make_new_context (poa,
                                                            TAO_ROOT_NAMING_CONTEXT,
                                                            context_size);

        }

#if !defined (CORBA_E_MICRO)
      // Register with the ORB's resolve_initial_references()
      // mechanism.  Primarily useful for dynamically loaded Name
      // Services.
      orb->register_initial_reference ("NameService",
                                       this->naming_context_.in ());
#endif /* CORBA_E_MICRO */

      // Set the ior of the root Naming Context.
      this->naming_service_ior_=
        orb->object_to_string (this->naming_context_.in ());

bool
LiveEntry::validate_ping (bool &want_reping, ACE_Time_Value& next)
{
  if (this->liveliness_ == LS_PING_AWAY ||
      this->liveliness_ == LS_DEAD ||
      this->listeners_.size () == 0)
    {
      if (ImR_Locator_i::debug () > 4)
        {
          ORBSVCS_DEBUG ((LM_DEBUG,
                          ACE_TEXT ("(%P|%t) LiveEntry::validate_ping, status ")
                          ACE_TEXT ("= %s, listeners = %d server %C\n"),
                          status_name (this->liveliness_), this->listeners_.size (),
                          this->server_.c_str()));
        }
      return false;
    }
  ACE_Time_Value now (ACE_High_Res_Timer::gettimeofday_hr());
  ACE_Time_Value diff = this->next_check_ - now;
  long msec = diff.msec();
  if (msec > 0)
    {
      if (!want_reping || this->next_check_ < next)
        {
          want_reping = true;
          next = this->next_check_;
        }
      if (ImR_Locator_i::debug () > 2)
        {
          ORBSVCS_DEBUG ((LM_DEBUG,
                          ACE_TEXT ("(%P|%t) LiveEntry::validate_ping, ")
                          ACE_TEXT ("status = %s, listeners = %d, ")
                          ACE_TEXT ("diff = %d,%d, msec = %d ")
                          ACE_TEXT ("server %C\n"),
                          status_name (this->liveliness_), this->listeners_.size (),
                          diff.sec(), diff.usec(), msec,
                          this->server_.c_str()));
        }
      return false;
    }
  switch (this->liveliness_)
    {
    case LS_UNKNOWN:
      break;
    case LS_ALIVE:
    case LS_TIMEDOUT:
      {
        ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, mon, this->lock_, false);
        this->next_check_ = now + owner_->ping_interval();
      }
      break;
    case LS_TRANSIENT:
    case LS_LAST_TRANSIENT:
      {
        int ms = this->next_reping ();
        if (ms != -1)
          {
            ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, mon, this->lock_, false);
            if (this->liveliness_ == LS_LAST_TRANSIENT)
              {
                this->liveliness_ = LS_TRANSIENT;
              }
            ACE_Time_Value next (ms / 1000, (ms % 1000) * 1000);
            this->next_check_ = now + next;
            if (ImR_Locator_i::debug () > 4)
              {
                ORBSVCS_DEBUG ((LM_DEBUG,
                                ACE_TEXT ("(%P|%t) LiveEntry::validate_ping, ")
                                ACE_TEXT ("transient, reping in %d ms, ")
                                ACE_TEXT ("server %C\n"),
                                ms, this->server_.c_str()));
              }
          }
        else
          {
            if (this->liveliness_ == LS_TRANSIENT)
              {
                ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, mon, this->lock_, false);
                this->liveliness_ = LS_LAST_TRANSIENT;
              }
            if (ImR_Locator_i::debug () > 2)
              {
                ORBSVCS_DEBUG ((LM_DEBUG,
                                ACE_TEXT ("(%P|%t) LiveEntry::validate_ping, ")
                                ACE_TEXT ("transient, no more repings, ")
                                ACE_TEXT ("server %C\n"),
                                this->server_.c_str()));
              }
            if (this->listeners_.size() > 0)
              {
                this->update_listeners ();
              }
            return false;
          }
      }
      break;
    default:;
    }
  return true;
}

void
TAO_EC_Default_ProxyPushSupplier::connect_push_consumer (
      RtecEventComm::PushConsumer_ptr push_consumer,
      const RtecEventChannelAdmin::ConsumerQOS& qos)
{
  // Nil PushConsumers are illegal
  if (CORBA::is_nil (push_consumer))
    throw CORBA::BAD_PARAM ();

  {
    ACE_GUARD_THROW_EX (
        ACE_Lock, ace_mon, *this->lock_,
        CORBA::INTERNAL ());
    // @@ RtecEventChannelAdmin::EventChannel::SYNCHRONIZATION_ERROR ());

    if (this->is_connected_i ())
      {
        if (this->event_channel_->consumer_reconnect () == 0)
          throw RtecEventChannelAdmin::AlreadyConnected ();

        // Re-connections are allowed....
        this->cleanup_i ();

        this->consumer_ =
          RtecEventComm::PushConsumer::_duplicate (push_consumer);
        this->qos_ = qos;
        this->child_ =
          this->event_channel_->filter_builder ()->build (this,
                                                          this->qos_);

        this->adopt_child (this->child_);

        TAO_EC_Unlock reverse_lock (*this->lock_);

        {
          ACE_GUARD_THROW_EX (
              TAO_EC_Unlock, ace_mon, reverse_lock,
              CORBA::INTERNAL ());
          // @@ RtecEventChannelAdmin::EventChannel::SYNCHRONIZATION_ERROR ());

          this->event_channel_->reconnected (this);
        }

        // A separate thread could have connected simultaneously,
        // this is probably an application error, handle it as
        // gracefully as possible
        if (this->is_connected_i ())
          return; // @@ Should we throw
      }

#if (TAO_HAS_CORBA_MESSAGING == 1)
      if ( consumer_validate_connection_ == 1 )
      {
        // Validate connection during connect.
        CORBA::PolicyList_var unused;
        int status = push_consumer->_validate_connection (unused);
#if TAO_EC_ENABLE_DEBUG_MESSAGES
        ORBSVCS_DEBUG ((LM_DEBUG, "Validated connection to PushConsumer on connect. Status[%d]\n", status));
#else
        ACE_UNUSED_ARG(status);
#endif /* TAO_EC_ENABLED_DEBUG_MESSAGES */
      }
#endif /* TAO_HAS_CORBA_MESSAGING == 1 */

      this->consumer_ =
        RtecEventComm::PushConsumer::_duplicate (push_consumer);
      this->qos_ = qos;

#if TAO_EC_ENABLE_DEBUG_MESSAGES
    ORBSVCS_DEBUG ((LM_DEBUG,
                "Building filters for consumer <%x>.\n",
                this));
#endif /* TAO_EC_ENABLED_DEBUG_MESSAGES */
    this->child_ =
      this->event_channel_->filter_builder ()->build (this,
                                                      this->qos_);

    this->adopt_child (this->child_);
  }

  // Notify the event channel...
  this->event_channel_->connected (this);
}

int
TAO_UIPMC_Protocol_Factory::init (int argc,
                                  ACE_TCHAR* argv [])
{
  ACE_Arg_Shifter arg_shifter (argc, argv);

  while (arg_shifter.is_anything_left ())
    {
      const ACE_TCHAR *current_arg = 0;

      // This option lets you override the default
      if (0 != (current_arg = arg_shifter.get_the_parameter
                (ACE_TEXT("-ORBListenOnAll"))))
        {
          this->listen_on_all_ = !!ACE_OS::atoi (current_arg);
          arg_shifter.consume_arg ();
        }
      else if (0 != (current_arg = arg_shifter.get_the_parameter
                (ACE_TEXT("-ORBListenerInterfaces"))) ||
               0 != (current_arg = arg_shifter.get_the_parameter
                (ACE_TEXT("-ORBListenerInterface"))) ||
               0 != (current_arg = arg_shifter.get_the_parameter
                (ACE_TEXT("-ORBListenInterfaces"))) ||
               0 != (current_arg = arg_shifter.get_the_parameter
                (ACE_TEXT("-ORBListenInterface"))) )
        {
          if (0 == ACE_OS::strcasecmp (current_arg, ACE_TEXT ("CopyPreferredInterfaces")) ||
              0 == ACE_OS::strcasecmp (current_arg, ACE_TEXT ("CopyPreferredInterface"))    )
            {
              // Add a special token (see TAO_UIPMC_Acceptor) which will be
              // subsituted (when we have access to the orb->params).
              if (this->listener_interfaces_.length ())
                this->listener_interfaces_+= ',';
              this->listener_interfaces_+= CopyPreferredInterfaceToken;
            }
          else
            {
              const char *const always_char = ACE_TEXT_ALWAYS_CHAR (current_arg);
              if (TAO_ORB_Parameters::check_preferred_interfaces_string (always_char))
                {
                  // Append any valid string to those already specified
                  // (by other -ORBListenerInterfaces options that have been
                  // seen previously)
                  if (this->listener_interfaces_.length ())
                    this->listener_interfaces_+= ',';
                  this->listener_interfaces_+= always_char;
                }
              else
                ORBSVCS_DEBUG ((LM_DEBUG,
                                ACE_TEXT ("TAO_UIPMC_Protocol_Factory::init - ")
                                ACE_TEXT ("ignoring invalid -ORBListenerInterfaces %C\n"),
                                always_char));
            }

          arg_shifter.consume_arg ();
        }
      else
        {
          ORBSVCS_DEBUG ((LM_DEBUG,
                      ACE_TEXT ("TAO_UIPMC_Protocol_Factory::init - ")
                      ACE_TEXT ("ignoring unknown option <%s>\n"),
                      arg_shifter.get_current ()));
          arg_shifter.ignore_arg ();
        }
    }

  return 0;
}

// FUZZ: disable check_for_ACE_Guard
bool
TAO_Notify_SequencePushConsumer::dispatch_from_queue (Request_Queue& requests, ACE_Guard <TAO_SYNCH_MUTEX> & ace_mon)
// FUZZ: enable check_for_ACE_Guard
{
  bool result = true;
  if (DEBUG_LEVEL > 0)
  {
    ORBSVCS_DEBUG ( (LM_DEBUG,
      ACE_TEXT ("(%P|%t) SequencePushConsumer dispatch queued requests. queue size:%u\n"),
      requests.size ()));
  }

  CORBA::ULong queue_size = ACE_Utils::truncate_cast<CORBA::ULong> (requests.size ());
  CORBA::Long max_batch_size = queue_size;
  if (this->max_batch_size_.is_valid () )
  {
    max_batch_size = this->max_batch_size_.value ();
  }
  CORBA::Long batch_size = queue_size;
  if (batch_size > max_batch_size)
  {
    batch_size = max_batch_size;
  }
  if (batch_size > 0)
  {
    CosNotification::EventBatch batch (batch_size);
    batch.length (batch_size);

    Request_Queue completed;

    CORBA::Long pos = 0;
    TAO_Notify_Method_Request_Event_Queueable * request = 0;
    while (pos < batch_size && requests.dequeue_head (request) == 0)
    {
      if (DEBUG_LEVEL > 0)
      {
        ORBSVCS_DEBUG ( (LM_DEBUG,
          ACE_TEXT ("(%P|%t) Sequence Dispatch Method_Request_Dispatch @%@\n"),
          request));
      }

      const TAO_Notify_Event * ev = request->event ();
      ev->convert (batch [pos]);
      ++pos;

      // note enqueue at head, use queue as stack.
      completed.enqueue_head (request);
    }
    batch.length (pos);
    ACE_ASSERT (pos > 0);

    ace_mon.release ();
    bool from_timeout = false;
    TAO_Notify_Consumer::DispatchStatus status =
      this->dispatch_batch (batch);
    ace_mon.acquire ();
    switch (status)
    {
    case DISPATCH_SUCCESS:
      {
        TAO_Notify_Method_Request_Event_Queueable * request = 0;
        while (completed.dequeue_head (request) == 0)
        {
          request->complete ();
          request->release ();
        }
        result = true;
        break;
      }
    case DISPATCH_FAIL_TIMEOUT:
      from_timeout = true;
      // Fall through
    case DISPATCH_FAIL:
      {
        TAO_Notify_Method_Request_Event_Queueable * request = 0;
        while (completed.dequeue_head (request) == 0)
        {
          if (request->should_retry ())
          {
            if (DEBUG_LEVEL > 0)
              ORBSVCS_DEBUG ((LM_DEBUG, ACE_TEXT ("(%P|%t) Consumer %d: Will retry %d\n"),
                          static_cast <int> (this->proxy ()->id ()),
                          request->sequence ()));
            requests.enqueue_head (request);
            result = false;
          }
          else
          {
            if (DEBUG_LEVEL > 0)
              ORBSVCS_DEBUG ((LM_DEBUG, ACE_TEXT ("(%P|%t) Consumer %d: Discarding %d\n"),
                          static_cast<int> (this->proxy ()->id ()),
                          request->sequence ()));
            request->complete ();
            request->release ();
          }
        }
        while (requests.dequeue_head (request) == 0)
        {
          if (request->should_retry ())
//.........这里部分代码省略.........

//.........这里部分代码省略.........
    CosNotifyComm::SequencePushSupplier_var stubSeqPushSupplier =
        CosNotifyComm::SequencePushSupplier::_nil();

    this->sequence_proxy_push_consumer_->connect_sequence_push_supplier (stubSeqPushSupplier.in());
    ///////////////////////////
    // Consumer registration

    // find the channel administrator for consumers
    this->consumer_admin_ = this->notify_channel_->default_consumer_admin ();
    if (CORBA::is_nil (this->consumer_admin_.in ()))
    {
        ORBSVCS_ERROR ((LM_ERROR,
                        "%T %n (%P|%t) NIL consumer admin\n"
                       ));
        result = -1;
    }
    // everything else happens when subscriber shows up

    ///////////////////////////////
    // Register with ReplicationManager
    if (this->rm_register_)
    {
        try
        {
            CORBA::Object_var rm_obj = orb->resolve_initial_references("ReplicationManager");
            this->replication_manager_ = ::FT::ReplicationManager::_narrow(rm_obj.in());
            if (!CORBA::is_nil (replication_manager_.in ()))
            {
                // @@: should we check to see if there's already one registered?
                FT::FaultNotifier_var notifier = FT::FaultNotifier::_narrow (this_obj.in ());
                if (! CORBA::is_nil (notifier.in ()))
                {
                    this->replication_manager_->register_fault_notifier(notifier.in ());
                    ORBSVCS_DEBUG ((LM_DEBUG,
                                    "FaultNotifier registered with ReplicationManager.\n"
                                   ));
                    this->registered_ = 1;
                }
                else
                {
                    ORBSVCS_ERROR ((LM_ERROR,
                                    "Error: Registration failed.  This is not a FaultNotifier (should not occur.)\n"
                                   ));
                }
            }
            else
            {
                ORBSVCS_ERROR ((LM_ERROR,"FaultNotifier: Can't resolve ReplicationManager, It will not be registered.\n" ));
            }
        }
        catch (const CORBA::Exception& ex)
        {
            ex._tao_print_exception (
                "FaultNotifier: Exception resolving ReplicationManager.  Notifier will not be registered.\n");
        }
    }
    else
    {
        ORBSVCS_DEBUG ((LM_DEBUG,
                        "FaultNotifier: ReplicationManager registration disabled.\n"
                       ));
    }
    ///////////////////////////////
    // Set up and ready for action
    // publish our IOR

DsLogAdmin::RecordList*
TAO_Hash_LogRecordStore::query_i (const char *constraint,
                                  DsLogAdmin::Iterator_out &iter_out,
                                  CORBA::ULong how_many)
{
  // Use an Interpreter to build an expression tree.
  TAO_Log_Constraint_Interpreter interpreter (constraint);

  // Sequentially iterate over all the records and pick the ones that
  // meet the constraints.

  // Allocate the list of <how_many> length.
  DsLogAdmin::RecordList* rec_list;
  ACE_NEW_THROW_EX (rec_list,
                    DsLogAdmin::RecordList (how_many),
                    CORBA::NO_MEMORY ());
  rec_list->length(how_many);

  // Create iterators
  LOG_RECORD_STORE_ITER iter (rec_map_.begin ());
  LOG_RECORD_STORE_ITER iter_end (rec_map_.end ());

  CORBA::ULong count = 0;       // count of matches found.

  for ( ; ((iter != iter_end) && (count < how_many)); ++iter)
    {
      // Use an evaluator.
      TAO_Log_Constraint_Visitor evaluator (iter->item ());

      // Does it match the constraint?
      if (interpreter.evaluate (evaluator) == 1)
        {
          if (TAO_debug_level > 0)
            {
              ORBSVCS_DEBUG ((LM_DEBUG,"Matched constraint! d = %Q, Time = %Q\n",
                iter->item ().id,
                iter->item ().time));
            }

        (*rec_list)[count] = iter->item ();
        // copy the log record.
        ++count;
      }
    }

  rec_list->length (count);

  if (iter != iter_end)         // There are more records to process.
    {
      // Create an iterator to pass out.
      TAO_Hash_Iterator_i *iter_query = 0;
      ACE_NEW_THROW_EX (iter_query,
                        TAO_Hash_Iterator_i (this->iterator_poa_.in (),
                                             this->reactor_,
                                             this,
                                             iter,
                                             iter_end,
                                             count,
                                             constraint,
                                             this->max_rec_list_len_),
                        CORBA::NO_MEMORY ());

      // Transfer ownership to the POA.
      PortableServer::ServantBase_var safe_iter_query = iter_query;

      // Activate it.
      PortableServer::ObjectId_var oid =
        this->iterator_poa_->activate_object (iter_query);
      CORBA::Object_var obj =
        this->iterator_poa_->id_to_reference (oid.in ());

      // Narrow it
      iter_out = DsLogAdmin::Iterator::_narrow (obj.in ());
    }

  return rec_list;
}

int
TAO::SSLIOP::Acceptor::ssliop_open_i (TAO_ORB_Core *orb_core,
                                      const ACE_INET_Addr& addr,
                                      ACE_Reactor *reactor)
{
  this->orb_core_ = orb_core;

  ACE_NEW_RETURN (this->creation_strategy_,
                  CREATION_STRATEGY (this->orb_core_),
                  -1);

  ACE_NEW_RETURN (this->concurrency_strategy_,
                  CONCURRENCY_STRATEGY (this->orb_core_),
                  -1);

  ACE_NEW_RETURN (this->accept_strategy_,
                  ACCEPT_STRATEGY (this->orb_core_,
                                   this->timeout_),
                  -1);

  u_short requested_port = addr.get_port_number ();
  if (requested_port == 0)
    {
      // don't care, i.e., let the OS choose an ephemeral port
      if (this->ssl_acceptor_.open (addr,
                                    reactor,
                                    this->creation_strategy_,
                                    this->accept_strategy_,
                                    this->concurrency_strategy_,
                                    0, 0, 0, 1,
                                    this->reuse_addr_) == -1)
        {
          if (TAO_debug_level > 0)
            ORBSVCS_DEBUG ((LM_DEBUG,
                        ACE_TEXT ("\n\nTAO (%P|%t) ")
                        ACE_TEXT ("SSLIOP_Acceptor::open_i - %p\n\n"),
                        ACE_TEXT ("cannot open acceptor")));
          return -1;
        }
    }
  else
    {
      ACE_INET_Addr a(addr);

      int found_a_port = 0;
      ACE_UINT32 last_port = requested_port + this->port_span_ - 1;
      if (last_port > ACE_MAX_DEFAULT_PORT)
        {
          last_port = ACE_MAX_DEFAULT_PORT;
        }

      for (ACE_UINT32 p = requested_port; p <= last_port; p++)
        {
          if (TAO_debug_level > 5)
            ORBSVCS_DEBUG ((LM_DEBUG,
                        ACE_TEXT ("TAO (%P|%t) IIOP_Acceptor::open_i() ")
                        ACE_TEXT ("trying to listen on port %d\n"), p));

          // Now try to actually open on that port
          a.set_port_number ((u_short)p);
          if (this->ssl_acceptor_.open (a,
                                        reactor,
                                        this->creation_strategy_,
                                        this->accept_strategy_,
                                        this->concurrency_strategy_,
                                        0, 0, 0, 1,
                                        this->reuse_addr_) != -1)
            {
              found_a_port = 1;
              break;
            }
        }

      // Now, if we couldn't locate a port, we punt
      if (! found_a_port)
        {
          if (TAO_debug_level > 0)
            ORBSVCS_DEBUG ((LM_DEBUG,
                        ACE_TEXT ("\n\nTAO (%P|%t) ")
                        ACE_TEXT ("SSLIOP_Acceptor::open_i - %p\n\n"),
                        ACE_TEXT ("cannot open acceptor")));
          return -1;
        }
    }

  ACE_INET_Addr ssl_address;

  // We do this to mak