pg_object_group.cpp

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

void TAO::PG_Object_Group::distribute_iogr (ACE_ENV_SINGLE_ARG_DECL)
{
  // assume internals is locked
  CORBA::String_var iogr =
    this->orb_->object_to_string (this->reference_.in()
                                  ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

//  size_t n_rep = 0; // for dump_ior below
  for ( MemberMap_Iterator it = this->members_.begin();
    it != this->members_.end();
    ++it)
  {
    MemberInfo const * info = (*it).int_id_;
    //
    // Unchecked narrow means the member doesn't have to actually implement the TAO_UpdateObjectGroup interface
    // PortableGroup::TAO_UpdateObjectGroup_var uog = PortableGroup::TAO_UpdateObjectGroup::_unchecked_narrow ( info->member_);
    // but it doesn work: error message at replica is:
    // TAO-FT (2996|976) - Wrong version information within the interceptor [1 | 0]
    // TAO_Perfect_Hash_OpTable:find for operation 'tao_update_object_group' (length=23) failed
    // back to using _narrow
    PortableGroup::TAO_UpdateObjectGroup_var uog =
      PortableGroup::TAO_UpdateObjectGroup::_narrow ( info->member_.in ());
    if (! CORBA::is_nil (uog.in ()) )
    {
      ACE_TRY_NEW_ENV
      {
        if (TAO_debug_level > 3)
        {
          ACE_DEBUG ((LM_DEBUG,
            "PG (%P|%t) -  Object_Group pushing IOGR to %s member: %s@%s.\n",
            (info->is_primary_ ? "Primary" : "Backup"),
            this->role_.c_str(),
            ACE_static_cast(const char *, info->location_[0].id)
            ));
        }
//        dump_ior ("group", "iogr", this->tagged_component_.object_group_ref_version, iogr);
//        CORBA::String_var replica_ior = this->orb_->object_to_string(uog.in() ACE_ENV_ARG_PARAMETER);
//        dump_ior (info->location_[0].id, "ior", (this->tagged_component_.object_group_ref_version * 100) + n_rep++, replica_ior);
        uog->tao_update_object_group (iogr.in (),
                                      this->tagged_component_.object_group_ref_version,
                                      info->is_primary_
                                      ACE_ENV_ARG_PARAMETER);
        ACE_TRY_CHECK;

      }
      ACE_CATCHANY
      {
        // we expect an exception
        // tao_update_object_group is not a real method
      }
      ACE_ENDTRY;
    }
    else
    {
      ACE_ERROR ((LM_ERROR,
        "TAO::PG_Object_Group::distribute iogr can't narrow member reference to PortableGroup::TAO_UpdateObjectGroup.\n"
        ));
    }
  }
}

PortableGroup::Locations * TAO::PG_Object_Group::locations_of_members (ACE_ENV_SINGLE_ARG_DECL)
  ACE_THROW_SPEC ((CORBA::SystemException))
{
  InternalGuard guard(this->internals_);
  PortableGroup::Locations * result = 0;

  size_t count = this->members_.current_size ();

  ACE_NEW_THROW_EX (
    result,
    PortableGroup::Locations (count),
    CORBA::NO_MEMORY() );
  ACE_CHECK_RETURN (0);

  result->length (count);

  size_t pos = 0;
  for (MemberMap_Iterator it = this->members_.begin();
      it != this->members_.end();
      this->members_.begin())
  {
    const PortableGroup::Location & location = (*it).ext_id_;
    PortableGroup::Location & out = (*result)[pos];
    out = location;
  }
  return result;
}

CORBA::Object_ptr TAO::PG_Object_Group::get_member_reference (
    const PortableGroup::Location & the_location
    ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ((
    CORBA::SystemException,
    PortableGroup::MemberNotFound))
{
  InternalGuard guard(this->internals_);
  CORBA::Object_var result = CORBA::Object::_nil ();

  MemberInfo * info;
  if (this->members_.find (the_location, info) == 0)
  {
    result = CORBA::Object::_duplicate(info->member_.in ());
  }
  else
  {
    ACE_THROW_RETURN (PortableGroup::MemberNotFound(), result._retn ());
  }
  return result._retn ();
}


PortableGroup::MembershipStyleValue TAO::PG_Object_Group::get_membership_style () const
{
  PortableGroup::MembershipStyleValue membership_style = 0;
  if (! TAO::find (properties_, PortableGroup::PG_MEMBERSHIP_STYLE, membership_style))
  {
    membership_style = TAO_PG_MEMBERSHIP_STYLE;
  }
  return membership_style;
}


PortableGroup::MinimumNumberMembersValue TAO::PG_Object_Group::get_minimum_number_members () const
{
  PortableGroup::MinimumNumberMembersValue minimum_number_members = 0;
  if (! TAO::find (properties_, PortableGroup::PG_MINIMUM_NUMBER_MEMBERS, minimum_number_members))
  {
    minimum_number_members = TAO_PG_MINIMUM_NUMBER_MEMBERS;
  }
  return minimum_number_members;
}

PortableGroup::InitialNumberMembersValue TAO::PG_Object_Group::get_initial_number_members () const
{
  PortableGroup::InitialNumberMembersValue initial_number_members = 0;
  if (! TAO::find (properties_, PortableGroup::PG_INITIAL_NUMBER_MEMBERS, initial_number_members))
  {
    initial_number_members = TAO_PG_INITIAL_NUMBER_MEMBERS;
  }
  return initial_number_members;
}

void TAO::PG_Object_Group::create_member (
    const PortableGroup::Location & the_location,
    const char * type_id,
    const PortableGroup::Criteria & the_criteria
    ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ( (CORBA::SystemException,
    PortableGroup::MemberAlreadyPresent,
    PortableGroup::NoFactory,
    PortableGroup::ObjectNotCreated,
    PortableGroup::InvalidCriteria,
    PortableGroup::CannotMeetCriteria))
{
  InternalGuard guard(this->internals_);

  if (0 != this->members_.find (the_location))
  {
    // @@ what if factories were passed as criteria?

    CORBA::String_var factory_type;
    PortableGroup::FactoryInfos_var factories =
    this->factory_registry_->list_factories_by_role (
          role_.c_str(),
          factory_type.out ()
          ACE_ENV_ARG_PARAMETER);
    ACE_CHECK;

    // @@ what if factory_type != type_id != this->type_id_

    int created = 0; // bool
    CORBA::ULong factory_count = factories->length ();
    for (CORBA::ULong factory_pos = 0;
       ! created && factory_pos < factory_count;
       ++factory_pos)
    {
      const PortableGroup::FactoryInfo & factory_info = (*factories)[factory_pos];
      if (factory_info.the_location == the_location)
      {
        // @@ should we merge the_criteria with factory_info.the_criteria?

        PortableGroup::GenericFactory::FactoryCreationId_var fcid;
        CORBA::Object_var member = factory_info.the_factory->create_object (
          type_id,
          the_criteria,
          fcid. out()
          ACE_ENV_ARG_PARAMETER);
        ACE_CHECK;

        // convert the new member to a stringified IOR to avoid contamination with group info
        CORBA::String_var member_ior_string =
          orb_->object_to_string (member.in ()
                                  ACE_ENV_ARG_PARAMETER);
        ACE_CHECK;

        PortableGroup::ObjectGroup_var new_reference =
          this->add_member_to_iogr (member.in ()
          ACE_ENV_ARG_PARAMETER);
        ACE_CHECK;

        // Convert new member back to a (non group) ior.
        CORBA::Object_var member_ior =
          this->orb_->string_to_object (member_ior_string.in ()
                                        ACE_ENV_ARG_PARAMETER);
        ACE_CHECK;

        MemberInfo * info = 0;
        ACE_NEW_THROW_EX (info, MemberInfo(
            member_ior.in(),
            the_location,
            factory_info.the_factory,
            fcid.in ()),
          CORBA::NO_MEMORY());
        ACE_CHECK;

        if (this->members_.bind (the_location, info) != 0)
        {
          ACE_THROW(CORBA::NO_MEMORY());
        }

        this->reference_ = new_reference; // note var-to-var assignment does a duplicate
        if (this->increment_version ())
        {
          this->distribute_iogr (ACE_ENV_SINGLE_ARG_PARAMETER);
          ACE_CHECK;
        }
        created = 1;
      }
    }
    if (! created)
    {
      ACE_THROW (PortableGroup::NoFactory ());
    }
  }
  else
  {
    ACE_THROW (PortableGroup::MemberAlreadyPresent ());
  }

}

void TAO::PG_Object_Group::create_members (size_t count ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ((
    CORBA::SystemException,
    PortableGroup::NoFactory
    ))
{
  // assume internals is locked
  // @@ what if factories were passed as criteria?

  CORBA::String_var factory_type;
  PortableGroup::FactoryInfos_var factories =
  this->factory_registry_->list_factories_by_role (
        role_.c_str(),
        factory_type.out ()
        ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  CORBA::ULong factory_count = factories->length ();
  if (factory_count > 0)
  {
    CORBA::ULong factory_pos = 0;
    while (members_.current_size () < count && factory_pos < factory_count)
    {
      const PortableGroup::FactoryInfo & factory_info = (*factories)[factory_pos];
      const PortableGroup::Location & factory_location = factory_info.the_location;
      if (0 != this->members_.find (factory_location))
      {
        ///////////////////////////////////////////
        // If a factory refuses to create a replica
        // it's not fatal.
        ACE_TRY_NEW_ENV
        {
          PortableGroup::GenericFactory::FactoryCreationId_var fcid;
          CORBA::Object_var member = factory_info.the_factory->create_object (
            this->type_id_.in (),
            factory_info.the_criteria,
            fcid. out()
            ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK;

          // convert the new member to a stringified IOR to avoid
          // contamination with group info
          CORBA::String_var member_ior_string =
            orb_->object_to_string (member.in ()
                                    ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK;

          PortableGroup::ObjectGroup_var new_reference =
            this->add_member_to_iogr (member.in ()
                                      ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK;

          // Convert new member back to a (non group) ior.
          CORBA::Object_var member_ior =
            this->orb_->string_to_object (member_ior_string.in ()
                                          ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK;

          MemberInfo * info = 0;
          ACE_NEW_THROW_EX (info, MemberInfo(
              member_ior.in(),
              factory_location,
              factory_info.the_factory,
              fcid.in ()),
            CORBA::NO_MEMORY());
          ACE_TRY_CHECK;

          if (this->members_.bind (factory_location, info) != 0)
          {
            ACE_TRY_THROW(CORBA::NO_MEMORY());
          }
          this->reference_ = new_reference; // note var-to-var assignment does a duplicate
        }
        ACE_CATCHANY
        {
          // log, but otherwise ignore the errorf
          if (TAO_debug_level > 0)
          {
            ACE_ERROR ((LM_ERROR,
              ACE_TEXT ("PG (%P|%t) Replica Factory @ %s refused create_object request for type %s\n"),
              ACE_static_cast (const char *, factory_info.the_location[0].id),
              ACE_static_cast (const char *, this->type_id_.in ())
              ));
          }
        }
        ACE_ENDTRY;
      }
    }

    if (this->increment_version ())
    {
      this->distribute_iogr (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_CHECK;
    }
  }
  else
  {
    ACE_THROW (PortableGroup::NoFactory());
  }
}

void TAO::PG_Object_Group::initial_populate (ACE_ENV_SINGLE_ARG_DECL)
{
  InternalGuard guard(this->internals_);
  if ( this->get_membership_style() == PortableGroup::MEMB_INF_CTRL )
  {
    PortableGroup::InitialNumberMembersValue initial_number_members = this->get_initial_number_members ();
    if (members_.current_size () < initial_number_members)
    {
      this->create_members (initial_number_members ACE_ENV_ARG_PARAMETER);
    }
  }
}

void TAO::PG_Object_Group::minimum_populate (ACE_ENV_SINGLE_ARG_DECL)
{
  InternalGuard guard(this->internals_);
  if ( this->get_membership_style() == PortableGroup::MEMB_INF_CTRL )
  {
    PortableGroup::MinimumNumberMembersValue minimum_number_members = this->get_minimum_number_members ();
    if (members_.current_size () < minimum_number_members)
    {
      this->create_members (minimum_number_members ACE_ENV_ARG_PARAMETER);
    }
  }
}

int TAO::PG_Object_Group::has_member_at (const PortableGroup::Location & location )
{
  return (0 == this->members_.find (location));
}




#if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION)

  template class ACE_Hash_Map_Manager_Ex <
    PortableGroup::Location,
    TAO::PG_Object_Group::MemberInfo *,
    TAO_PG_Location_Hash,
    TAO_PG_Location_Equal_To,
    TAO_SYNCH_MUTEX>;
  template class ACE_Hash_Map_Entry <
    PortableGroup::Location,
    TAO::PG_Object_Group::MemberInfo *>;
  template class ACE_Hash_Map_Iterator_Ex <
    PortableGroup::Location,
    TAO::PG_Object_Group::MemberInfo *,
    TAO_PG_Location_Hash,
    TAO_PG_Location_Equal_To,
    TAO_SYNCH_MUTEX>;

#elif defined (ACE_HAS_TEMPLATE_INSTANTIATION_PRAGMA)

# pragma instantiate ACE_Hash_Map_Manager_Ex <
    PortableGroup::Location,
    TAO::PG_Object_Group::MemberInfo *,
    TAO_PG_Location_Hash,
    TAO_PG_Location_Equal_To,
    TAO::PG_Object_Group::MemberMapMutex>
# pragma instantiate ACE_Hash_Map_Entry <PortableGroup::Location, TAO::PG_Object_Group::MemberInfo *>
# pragma instantiate ACE_Hash_Map_Iterator_Ex <
    PortableGroup::Location,
    TAO::PG_Object_Group::MemberInfo *,
    TAO_PG_Location_Hash,
    TAO_PG_Location_Equal_To,
    TAO::PG_Object_Group::MemberMapMutex>

#endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION */