---

CdbBdbSRegistryP.cc

Go to the documentation of this file.

// File and Version Information:
//      $Id: CdbBdbSRegistryP.cc,v 1.12 2004/12/20 20:20:04 gapon Exp $

/// Implementation file for the CdbBdbSRegistryP class
/**
  * @see CdbBdbSRegistryP
  */

#include "BaBar/BaBar.hh"

#include "CdbBdbShared/CdbBdbSRegistryP.hh"
#include "CdbBdbShared/CdbBdbSUtils.hh"
#include "CdbBdbShared/CdbBdbSOriginP.hh"
#include "CdbBdbShared/CdbBdbSSimpleClusterP.hh"
#include "CdbBdbShared/CdbBdbSViewP.hh"
#include "CdbBdbShared/CdbBdbSFolderP.hh"
#include "CdbBdbShared/CdbBdbSConditionAtFolderP.hh"
#include "CdbBdbShared/CdbBdbSConditionP.hh"

#include "CdbBase/CdbPathName.hh"

#include "BdbClustering/BdbCondIndexClusteringHint.hh"

#include <assert.h>
#include <stdio.h>

#include <string>
#include <vector>
using std::cout;
using std::endl;
using std::ostream;

static const char* databaseCoreName    = "con_cdb_database_";
static const char* masterDatabaseName  = "con_cdb_database";
static const char* systemContainerName = "System";
static const char* systemRegistryName  = "Registry";
//static const char* regularClusterName  = "initial";
static const char* mainViewName        = "main";

CdbBdbSRegistryP::CdbBdbSRegistryP( d_UShort       theOriginId,
                                    const char*    theDescription,
                                    const BdbTime& theCreationTime,
                                    const BdbTime& theId,
                                    bool           hasLocalCollectionsFlag ) :
    _originId(theOriginId),
    _description(theDescription),
    _created(theCreationTime),
    _id(theId),
    _hasLocalCollections(hasLocalCollectionsFlag),
    _conditionCollectionRef(0),
    _viewCollectionRef(0),
    _clusterCollectionRef(0),
    _conditionRallocatorRef(0),
    _viewRallocatorRef(0),
    _clusterRallocatorRef(0)
{
    assert( 0 != theDescription );
    assert(( BdbTime::minusInfinity != theCreationTime ) && ( BdbTime::plusInfinity != theCreationTime ));
    assert(( BdbTime::minusInfinity != theId )           && ( BdbTime::plusInfinity != theId ));

  // Create local collections and resources allocators if needed.
  //
  // NOTE: To minimize lock conflicts these collections can be created in
  //       other (than the current) containers.

    if( _hasLocalCollections ) {

        _conditionCollectionRef = new( ooThis( )) CdbBdbSConditionCollectionP( );
        _viewCollectionRef      = new( ooThis( )) CdbBdbSViewCollectionP     ( );
        _clusterCollectionRef   = new( ooThis( )) CdbBdbSClusterCollectionP  ( );

        _conditionRallocatorRef = new( ooThis( )) CdbBdbSRallocatorP< d_UShort >( );
        _viewRallocatorRef      = new( ooThis( )) CdbBdbSRallocatorP< d_UShort >( );
        _clusterRallocatorRef   = new( ooThis( )) CdbBdbSRallocatorP< d_UShort >( );

/**
 ** Do not create initial cluster. This should be done explicitly
 ** through a separate procedure.
 **

      // Create initial cluster for non-partitioned clustered conditions

        {
            CdbStatus result;

          // Create the data structures

            const d_UShort cId = 0;

            result = _clusterRallocatorRef->force( cId );
            assert( CdbStatus::Success == result );

            BdbRef(CdbBdbSClusterP) cRef = new( ooThis( )) CdbBdbSSimpleClusterP( regularClusterName,
                                                                                  cId,
                                                                                  "initial cluster for non-partitionable conditions",
                                                                                  theCreationTime );

            result = _clusterCollectionRef->insert( cRef );
            assert( CdbStatus::Success == result );

          // Obtain the local DBID range. We'll use it as part of
          // the database file name

            std::string localRangeName;

            result = CdbBdbSUtils::getLocalDbIdRange( localRangeName );
            assert( CdbStatus::Success == result );


          // Crate the database file and container for clustered conditions (their metadata
          // and condition objects).

            std::string databaseSuffix = std::string( "c0_i0_" ) + localRangeName;

            BdbCondIndexClusteringHint hint( "cdb",
                                             "System",
                                             databaseSuffix.c_str( ));

            BdbHandle(BdbContObj) contH;
            contH = hint.updatedHint( );
            assert( ! BdbIsNull(contH));
            assert( hint.newContainerCreated( ));
        }

**
** End of the inhibited initial cluster creation zone.
**/

      // Create an empty 'main' view.
      //
      // NOTE: This view should be properly loaded elsewhere.

        {
            CdbStatus result;

            const d_UShort vId = 0;

            result = _viewRallocatorRef->force( vId );
            assert( CdbStatus::Success == result );

          // Create new view object

            BdbRef(CdbBdbSViewP) viewRef =
                new( _viewCollectionRef ) CdbBdbSViewP( mainViewName,
                                                        vId,
                                                        "This is a special view. It was created automatically when initializing the current database",
                                                        theCreationTime,
                                                        BdbTime::minusInfinity,
                                                        BdbTime::plusInfinity,
                                                        0 );
            assert( ! BdbIsNull(viewRef));

          // Register this view in the collection

            result = _viewCollectionRef->insert( viewRef );
            assert( CdbStatus::Success == result );
        }
    }

  // Name self in the scope of its container

    BdbHandle(BdbContObj) cH;
    ooThis( ).containedIn( cH );

    BdbStatus status;
    status = ooThis( ).nameObj( cH,
                                systemRegistryName );
    assert( BdbcSuccess == status );
}

CdbBdbSRegistryP::~CdbBdbSRegistryP( )
{
    if( ! BdbIsNull(_conditionCollectionRef)) BdbDelete(_conditionCollectionRef);
    if( ! BdbIsNull(_viewCollectionRef))      BdbDelete(_viewCollectionRef);
    if( ! BdbIsNull(_clusterCollectionRef))   BdbDelete(_clusterCollectionRef);

    if( ! BdbIsNull(_conditionRallocatorRef)) BdbDelete(_conditionRallocatorRef);
    if( ! BdbIsNull(_viewRallocatorRef))      BdbDelete(_viewRallocatorRef);
    if( ! BdbIsNull(_clusterRallocatorRef))   BdbDelete(_clusterRallocatorRef);
}

const char*
CdbBdbSRegistryP::defaultDbIdRange( )
{
    return "default";
}

d_UShort
CdbBdbSRegistryP::originId( ) const
{
    return _originId;
}

ooString(32)
CdbBdbSRegistryP::description( ) const
{
    return _description;
}

BdbTime
CdbBdbSRegistryP::created( ) const
{
    return _created;
}

BdbTime
CdbBdbSRegistryP::id( ) const
{
    return _id;
}

bool
CdbBdbSRegistryP::hasLocalCollections( ) const
{
    return _hasLocalCollections;
}

BdbRef(CdbBdbSConditionCollectionP)
CdbBdbSRegistryP::conditionCollection( ) const
{
    return _conditionCollectionRef;
}

BdbRef(CdbBdbSViewCollectionP)
CdbBdbSRegistryP::viewCollection( ) const
{
    return _viewCollectionRef;
}

BdbRef(CdbBdbSClusterCollectionP)
CdbBdbSRegistryP::clusterCollection( ) const
{
    return _clusterCollectionRef;
}

BdbRef(CdbBdbSRallocatorP< d_UShort >)
CdbBdbSRegistryP::conditionRallocator( ) const
{
    return _conditionRallocatorRef;
}

BdbRef(CdbBdbSRallocatorP< d_UShort >)
CdbBdbSRegistryP::viewRallocator( ) const
{
    return _viewRallocatorRef;
}

BdbRef(CdbBdbSRallocatorP< d_UShort >)
CdbBdbSRegistryP::clusterRallocator( ) const
{
    return _clusterRallocatorRef;
}

void
CdbBdbSRegistryP::dump( ostream& o ) const
{
    o << "            ORIGIN ID: " << _originId << endl
      << "          DESCRIPTION: " << _description.head( ) << endl
      << "              CREATED: " << _created << endl
      << "                   ID: " << _id << endl
      << "            IS MASTER: " << ( isMaster( ) ? "Yes" : "No" ) << endl
      << "HAS LOCAL COLLECTIONS: " << ( _hasLocalCollections ? "Yes" : "No" ) << endl;
}

CdbStatus
CdbBdbSRegistryP::updateMainView( )
{
    const char* errorStr = "CdbBdbSRegistryP::updateMainView() - ERROR.";

    CdbStatus result = CdbStatus::Error;
    do {

      // Find the main view first

        BdbRef(CdbBdbSViewP) viewRef;
        {
            if( BdbIsNull(_viewCollectionRef)) {
                result = CdbStatus::NotFound;
                break;
            }
            if( CdbStatus::Success != _viewCollectionRef->find( mainViewName, viewRef )) {
                cout << errorStr << endl
                     << "    The main view does not exist in the local database." << endl
                     << "    The database may not be properly initialized/loaded." << endl;
                break;
            }
        }
        assert( !BdbIsNull(viewRef));

      // Make sure the view is not frozen.

        if( viewRef->isFrozen( )) {
            cout << errorStr << endl
                 << "    The main view is found to be in the 'frozen' state. This should never happen." << endl
                 << "    The database may not be properly initialized/loaded." << endl;
            break;
        }

      // Reset default configuration

        if( CdbStatus::Success != viewRef->setDefaultConfig( 0 )) {
            cout << errorStr << endl
                 << "    Failed to reset the default configuration at the main view." << endl;
            break;
        }

      // Remove everything from the view below the 'root' folder.
      //
      // NOTE: This algorithm is required to have two passes since we can't iterate over
      //       collection whose components are going to change. Otherwise it's going to be
      //       a crash inside Objectivity iterator's code.

        BdbRef(CdbBdbSFolderP) rootFolderRef = viewRef->rootFolder( );
        assert( !BdbIsNull(rootFolderRef));
        {
            CdbItr<const char*> fItr;
            if( CdbStatus::Success != rootFolderRef->folderIterator( fItr )) {
                cout << errorStr << endl
                     << "    Failed to set up an iterator of subfolders at the level of the 'root' folder." << endl;
                break;
            }
            std::vector<std::string> folders;
            while( fItr.next( )) {
                std::string fPathName = std::string( rootFolderRef->name( )) + std::string( fItr.value( ));
                folders.push_back( fPathName.c_str( ));
            }
            bool failed = false;
            for( std::vector<std::string>::iterator i = folders.begin( ); i != folders.end( ); ++i ) {

                bool forceRecursiveRemovalFlag = true;
                const char* f = (*i).c_str( );
                if( CdbStatus::Success != viewRef->removeFolder( f, forceRecursiveRemovalFlag )) {
                    cout << errorStr << endl
                         << "    Failed to remove folder: \"" << f << "\"" << endl;
                    failed = true;
                    break;
                }
            }
            if( failed ) break;
        }
        {
            CdbItr<const char*> cItr;
            if( CdbStatus::Success != rootFolderRef->conditionIterator( cItr )) {
                cout << errorStr << endl
                     << "    Failed to set up an iterator of conditions at the level of the 'root' folder." << endl;
                break;
            }
            std::vector<std::string> conditions;
            while( cItr.next( )) {
                const char* cName     = cItr.value( );
                std::string cPathName = std::string( rootFolderRef->name( )) + std::string( cName );
                conditions.push_back( cPathName.c_str( ));
            }
            bool failed = false;
            for( std::vector<std::string>::iterator i = conditions.begin( ); i != conditions.end( ); ++i ) {

                const char* c = (*i).c_str( );
                if( CdbStatus::Success != viewRef->removeCondition( c )) {
                    cout << errorStr << endl
                         << "    Failed to remove condition: \"" << c << "\"" << endl;
                    failed = true;
                    break;
                }
            }
            if( failed ) break;
        }

      // Iterate over:
      //
      //   <origin> -> { REGULAR | PARTITIONABLE } -> <cluster> -> <condition>
      //
      // And create the following branches of names at the view
      //
      //   /NAMES_ONLY/<origin_name>/REGULAR/<cluster_name>/<condition_name>
      //   /NAMES_ONLY/<origin_name>/PARTITIONABLE/<cluster_name>/<condition_name>
      //   /NAMES_AND_IDS/<origin_id>:<origin_name>/REGULAR/<cluster_id>:<cluster_name>/<condition_id>:<condition_name>
      //   /NAMES_AND_IDS/<origin_id>:<origin_name>/PARTITIONABLE/<cluster_id>:<cluster_name>/<condition_id>:<condition_name>
      //
      // NOTE: Put empty configurations into each found condition to prevent this view from
      //       any use by regular users. Remember, that the sole purpose of the view is to help
      //       managing the database.

        BdbTime creationTime( BdbTime::now( ));

        if( CdbStatus::Success != viewRef->createFolder( rootFolderRef->name( ),
                                                         "NAMES_ONLY",
                                                         "This folder begins a branch of names of persistent components only.",
                                                         creationTime )) {
            cout << errorStr << endl
                 << "    Failed to create new folder." << endl;
            break;
        }
        if( CdbStatus::Success != viewRef->createFolder( rootFolderRef->name( ),
                                                         "NAMES_AND_IDS",
                                                         "This folder begins a branch of both names and identifiers of persistent components.",
                                                          creationTime )) {
            cout << errorStr << endl
                 << "    Failed to create new folder." << endl;
            break;
        }
        std::string namesOnlyPath   = std::string( rootFolderRef->name( )) + std::string( "NAMES_ONLY" );
        std::string namesAndIdsPath = std::string( rootFolderRef->name( )) + std::string( "NAMES_AND_IDS" );

        BdbHandle(CdbBdbSRegistryP) masterRegistryH;
        if( CdbStatus::Success != findMaster( masterRegistryH )) {
            cout << errorStr << endl
                 << "    Failed to find the master registry." << endl
                 << "    The database may not be properly initialized/loaded." << endl;
            break;
        }

        BdbHandle(CdbBdbSOriginCollectionP) originCollectionH;
        originCollectionH = masterRegistryH->originCollection( );
        if( BdbIsNull(originCollectionH)) {
            cout << errorStr << endl
                 << "    Failed to find the collection of origins at the master registry." << endl
                 << "    The database may not be properly initialized/loaded." << endl;
            break;
        }

        bool failed = false;

        CdbBdbSOriginCollectionP::IteratorOfIdentifiers originIdItr = originCollectionH->iterator_identifiers( );
        while( originIdItr.next( )) {

            d_UShort originId = originIdItr.value( );

          // Check if the origin corresponding to that identifier is present in the federation.
          // If not then just skip the rest of the loop since some origins identifiers
          // from the current collection may be just pre-allocated.

            BdbRef(CdbBdbSOriginP) originRef;
            {
                CdbStatus s = originCollectionH->find( originId, originRef );
                if( CdbStatus::NotFound == s ) {

                  // There is no real origin for this id. Thsi is a normal
                  // case - just ignore this origin.

                    continue;

                } else if( CdbStatus::Success == s ) {


                  // Found.

                    ;

                } else {
                    cout << errorStr << endl
                         << "    Failed to find the origin for origin ID=" << originId << endl
                         << "    The database may not be properly initialized/loaded." << endl;
                    failed = true;
                    break;
                }
            }

          // Check if the registry corresponding to the above found origin is present in the federation.
          // If not then just skip the rest of the loop since this registry migh not be imported into
          // the current federation yet.

            BdbHandle(CdbBdbSRegistryP) registryH;
            {
                CdbStatus s = findByOrigin( registryH, originId );
                if( CdbStatus::NotFound == s ) {

                  // There is no real registry for the above found origin. Thsi is a normal
                  // case - just ignore this origin.

                    continue;

                } else if( CdbStatus::Success == s ) {


                  // Found.

                    ;

                } else {
                    cout << errorStr << endl
                         << "    Failed to find the registry for the origin ID=" << originId << endl
                         << "    The database may not be properly initialized/loaded." << endl;
                    failed = true;
                    break;
                }
            }

          // Create 'origin' level folders

            char originIdStr[12];
            sprintf( originIdStr, "%u", originId );

            std::string originNameOnly  = originRef->name( ).head( );
            std::string originNameAndId = std::string( originIdStr ) + std::string( ":" ) + originNameOnly;

            if( CdbStatus::Success != viewRef->createFolder( namesOnlyPath.c_str( ),
                                                             originNameOnly.c_str( ),
                                                             "This folder corresponds to an origin.",
                                                             creationTime )) {
                cout << errorStr << endl
                     << "    Failed to create new folder." << endl;
                failed = true;
                break;
            }
            if( CdbStatus::Success != viewRef->createFolder( namesAndIdsPath.c_str( ),
                                                             originNameAndId.c_str( ),
                                                             "This folder corresponds to an origin.",
                                                             creationTime )) {
                cout << errorStr << endl
                     << "    Failed to create new folder." << endl;
                failed = true;
                break;
            }
            std::string originNamesOnlyPath   = namesOnlyPath   + std::string( CdbPathName::separator( )) + originNameOnly;
            std::string originNamesAndIdsPath = namesAndIdsPath + std::string( CdbPathName::separator( )) + originNameAndId;

          // Proceed down to the current origin

            const char* conditionTypeName[2] = { "REGULAR", "PARTITIONABLE" };
            bool        isPartitionable  [2] = { false,     true };

            for( int i = 0; i < 2; ++i ) {

              // Get to an appropriate cluster.
              // Skip further actions if there is no such cluster at the current origin.

                BdbRef(CdbBdbSClusterCollectionP) clusterCollectionRef;
                {
                    if( isPartitionable[i] ) {
                        clusterCollectionRef = registryH->pClusterCollection( );
                    } else {
                        clusterCollectionRef = registryH->clusterCollection( );
                    }
                    if( BdbIsNull( clusterCollectionRef )) {

                      // This is a normal situation - some databases are not allowed local collections
                      // of clusters/conditions.

                        continue;
                    }
                }

              // Create condition type folders

                if( CdbStatus::Success != viewRef->createFolder( originNamesOnlyPath.c_str( ),
                                                                 conditionTypeName[i],
                                                                 "This folder corresponds to a certain type of conditions.",
                                                                 creationTime )) {
                    cout << errorStr << endl
                         << "    Failed to create new folder." << endl;
                    failed = true;
                    break;
                }
                if( CdbStatus::Success != viewRef->createFolder( originNamesAndIdsPath.c_str( ),
                                                                 conditionTypeName[i],
                                                                 "This folder corresponds to a certain type of conditions.",
                                                                 creationTime )) {
                    cout << errorStr << endl
                         << "    Failed to create new folder." << endl;
                    failed = true;
                    break;
                }
                std::string typeNamesOnlyPath   = originNamesOnlyPath   + std::string( CdbPathName::separator( )) + std::string( conditionTypeName[i] );
                std::string typeNamesAndIdsPath = originNamesAndIdsPath + std::string( CdbPathName::separator( )) + std::string( conditionTypeName[i] );

              // Proceed down with the collection of clusters by iterating over the identifiers/names
              // of clusters in that collection.

                CdbBdbSClusterCollectionP::IteratorOfIdentifiers clusterItr = clusterCollectionRef->iterator_identifiers( );
                while( clusterItr.next( )) {

                    d_UShort clusterId = clusterItr.value( );

                    BdbRef(CdbBdbSClusterP) clusterRef;
                    if( CdbStatus::Success != clusterCollectionRef->find( clusterId, clusterRef )) {
                        cout << errorStr << endl
                             << "    Failed to create new folder." << endl;
                        failed = true;
                        break;
                    }

                  // Create 'cluster' level folders

                    char clusterIdStr[12];
                    sprintf( clusterIdStr, "%u", clusterId );

                    std::string clusterNameOnly  = clusterRef->name( ).head( );
                    std::string clusterNameAndId = std::string( clusterIdStr ) + std::string( ":" ) + clusterNameOnly;

                    if( CdbStatus::Success != viewRef->createFolder( typeNamesOnlyPath.c_str( ),
                                                                     clusterNameOnly.c_str( ),
                                                                     "This folder corresponds to a cluster of conditions.",
                                                                     creationTime )) {
                        cout << errorStr << endl
                             << "    Failed to create new folder." << endl;
                        failed = true;
                        break;
                    }
                    if( CdbStatus::Success != viewRef->createFolder( typeNamesAndIdsPath.c_str( ),
                                                                     clusterNameAndId.c_str( ),
                                                                     "This folder corresponds to a cluster of conditions.",
                                                                     creationTime )) {
                        cout << errorStr << endl
                             << "    Failed to create new folder." << endl;
                        failed = true;
                        break;
                    }
                    std::string clusterNamesOnlyPath   = typeNamesOnlyPath   + std::string( CdbPathName::separator( )) + clusterNameOnly;
                    std::string clusterNamesAndIdsPath = typeNamesAndIdsPath + std::string( CdbPathName::separator( )) + clusterNameAndId;

                  // Proceed down to the current cluster, find its conditions and include them
                  // as conditions into teh view.

                    CdbItr< BdbRef(CdbBdbSConditionP)> conditionItr;
                    if( CdbStatus::Success != clusterRef->iterator( conditionItr )) {
                        cout << errorStr << endl
                             << "    Failed to set up an iterator of conditions at the current cluster." << endl;
                        failed = true;
                        break;
                    }
                    while( conditionItr.next( )) {

                        BdbRef(CdbBdbSConditionP) conditionRef = conditionItr.value( );

                        d_UShort conditionId = conditionRef->id( );

                        char conditionIdStr[12];
                        sprintf( conditionIdStr, "%u", conditionId );

                        std::string conditionNameOnly  = conditionRef->name( ).head( );
                        std::string conditionNameAndId = std::string( conditionIdStr ) + std::string( ":" ) + conditionNameOnly;

                        CdbBdbSId extendedConditionId( registryH->originId( ), conditionId );

                        if( CdbStatus::Success != viewRef->createCondition( clusterNamesOnlyPath.c_str( ),
                                                                            conditionNameOnly.c_str( ),
                                                                            "This condition was created automatically.",
                                                                            creationTime,
                                                                            extendedConditionId )) {
                            cout << errorStr << endl
                                 << "    Failed to create new condition." << endl;
                            failed = true;
                            break;
                        }
                        if( CdbStatus::Success != viewRef->createCondition( clusterNamesAndIdsPath.c_str( ),
                                                                            conditionNameAndId.c_str( ),
                                                                            "This condition was created automatically.",
                                                                            creationTime,
                                                                            extendedConditionId )) {
                            cout << errorStr << endl
                                 << "    Failed to create new condition." << endl;
                            failed = true;
                            break;
                        }

                      // End of the conditions loop
                    }
                    if( failed ) break;

                  // End of the clusters loop
                }
                if( failed ) break;

              // End of the condition types loop
            }
            if( failed ) break;

          // End of the origins loop
        }
        if( failed ) break;

      // Done

        result = CdbStatus::Success;

    } while( false );

    return result;
}

CdbStatus 
CdbBdbSRegistryP::findMaster( BdbHandle(CdbBdbSRegistryP)& theRegistryH,
                              const BdbMode                theOpenMode )
{
    CdbStatus result = CdbStatus::NotFound;
    do {

      // Find a container where the desired object is supposed to be

        BdbHandle(BdbContObj) cH;

        if( CdbStatus::Success != CdbBdbSUtils::findContainer( cH,
                                                               masterDatabaseName,
                                                               systemContainerName,
                                                               theOpenMode )) {
            break;
        }

      // Find the desired object

        BdbHandle(CdbBdbSRegistryP) registryH;

        BdbStatus status = registryH.lookupObj( cH,
                                                systemRegistryName,
                                                theOpenMode );
        if(( BdbcSuccess != status ) || BdbIsNull(registryH)) {
            break;
        }

      // Done

        theRegistryH = registryH;

        result = CdbStatus::Success;

    } while( false );

    return result;
}

CdbStatus
CdbBdbSRegistryP::findLocal( BdbHandle(CdbBdbSRegistryP)& theRegistryH,
                             const BdbMode                theOpenMode  )
{
    const char* errorStr = "CdbBdbSRegistryP::findLocal() - ERROR.";

    CdbStatus result = CdbStatus::NotFound;
    do {

      // Find the master registry first

        BdbHandle(CdbBdbSRegistryP) masterRegistryH;
        if( CdbStatus::Success != findMaster( masterRegistryH )) {
            break;
        }

      // Find the master DBID range

        BdbRef(CdbBdbSOriginCollectionP) originCollRef;
        originCollRef = masterRegistryH->originCollection( );
        assert( ! BdbIsNull(originCollRef));

        BdbRef( CdbBdbSOriginP ) originRef;
        if( CdbStatus::Success != originCollRef->find( masterRegistryH->originId( ),
                                                       originRef )) {
            cout << errorStr << endl
                 << "    Failed to obtain the origin information for the master registry." << endl
                 << "    The Condition/DB might be inproperly initialized or broken." << endl;
            result = CdbStatus::Error;
            break;
        }
        std::string masterRangeName;
        {
            ooString(32) dbidrange = originRef->dbidrange( );
            assert( 0 != dbidrange.head( ));
            masterRangeName = dbidrange.head( );
        }

      // Get the local DBID range Name

        std::string localRangeName;

        if( CdbStatus::Success != CdbBdbSUtils::getLocalDbIdRange( localRangeName )) {
            cout << errorStr << endl
                 << "    Failed to determine the local DBID range. This may be an indication" << endl
                 << "    that the Condition/DB federation has not been properly initialized." << endl;
            result = CdbStatus::Error;
            break;
        }

      // Trivial optimization in case if the local DBID range corresponds to
      // the master database.

        if( masterRangeName == localRangeName ) {
            theRegistryH = masterRegistryH;
        } else {

          // The above mentioned optimization does not work.

          // Try to locate the local registry container

            std::string localDatabaseName = databaseCoreName + localRangeName;

            BdbHandle(BdbContObj) cH;

            if( CdbStatus::Success != CdbBdbSUtils::findContainer( cH,
                                                                   localDatabaseName.c_str( ),
                                                                   systemContainerName,
                                                                   theOpenMode )) {
                break;
            }

          // Find the desired object

            BdbHandle(CdbBdbSRegistryP) localRegistryH;

            BdbStatus status = localRegistryH.lookupObj( cH,
                                                         systemRegistryName,
                                                         theOpenMode );
            if(( BdbcSuccess != status ) || BdbIsNull(localRegistryH)) {
                break;
            }
            theRegistryH = localRegistryH;
        }

      // Done

        result = CdbStatus::Success;

    } while( false );

    return result;
}

CdbStatus
CdbBdbSRegistryP::findByOrigin( BdbHandle(CdbBdbSRegistryP)& theRegistryH,
                                d_UShort                     theOriginId,
                                const BdbMode                theOpenMode )
{
    const char* errorStr = "CdbBdbSRegistryP::findByOrigin(id) - ERROR.";

    CdbStatus result = CdbStatus::NotFound;
    do {

      // Find the master registry first

        BdbHandle(CdbBdbSRegistryP) masterRegistryH;
        if( CdbStatus::Success != findMaster( masterRegistryH )) {
            break;
        }

      // Minor optimization in case if the master is what we're looking for

        if( theOriginId == masterRegistryH->originId( )) {
            theRegistryH = masterRegistryH;
        } else {

          // Find the DBID range corresponding to the specified
          // origin identifier

            BdbRef(CdbBdbSOriginCollectionP) originCollRef;
            originCollRef = masterRegistryH->originCollection( );
            assert( ! BdbIsNull(originCollRef));

            BdbRef( CdbBdbSOriginP ) originRef;
            if( CdbStatus::Success != originCollRef->find( theOriginId,
                                                           originRef )) {
                cout << errorStr << endl
                     << "    Failed to obtain the origin information for the registry." << endl
                     << "    The Condition/DB might be inproperly initialized or broken." << endl;
                result = CdbStatus::Error;
                break;
            }
            std::string rangeName;
            {
                ooString(32) dbidrange = originRef->dbidrange( );
                assert( 0 != dbidrange.head( ));
                rangeName = dbidrange.head( );
            }

          // Try to locate the registry container

            std::string databaseName = databaseCoreName + rangeName;

            BdbHandle(BdbContObj) cH;

            if( CdbStatus::Success != CdbBdbSUtils::findContainer( cH,
                                                                   databaseName.c_str( ),
                                                                   systemContainerName,
                                                                   theOpenMode )) {
                break;
            }

          // Find the desired object

            BdbHandle(CdbBdbSRegistryP) localRegistryH;

            BdbStatus status = localRegistryH.lookupObj( cH,
                                                         systemRegistryName,
                                                         theOpenMode );
            if(( BdbcSuccess != status ) || BdbIsNull(localRegistryH)) {
                break;
            }
            theRegistryH = localRegistryH;
        }

      // Done

        result = CdbStatus::Success;

    } while( false );

    return result;
}

CdbStatus
CdbBdbSRegistryP::findByOrigin( BdbHandle(CdbBdbSRegistryP)& theRegistryH,
                                const char*                  theOriginName,
                                const BdbMode                theOpenMode )
{
    const char* errorStr = "CdbBdbSRegistryP::findByOrigin(name) - ERROR.";

    CdbStatus result = CdbStatus::NotFound;
    do {

      // Verify parameters

        if( 0 == theOriginName ) {
            cout << "    Null pointer passed where the origin name was expected." << endl;
            break;
        }

      // Find the master registry first

        BdbHandle(CdbBdbSRegistryP) masterRegistryH;
        if( CdbStatus::Success != findMaster( masterRegistryH )) {
            break;
        }

      // Special handling for the MASTER since master's database file
      // is named in a special way.

        if( 0 == strcmp( theOriginName, CdbBdbSOriginP::type2name( CdbBdbSOriginP::MASTER ))) {
            theRegistryH = masterRegistryH;
        } else {

          // Find the DBID range corresponding to the specified
          // origin identifier

            BdbRef(CdbBdbSOriginCollectionP) originCollRef;
            originCollRef = masterRegistryH->originCollection( );
            assert( ! BdbIsNull(originCollRef));

            BdbRef( CdbBdbSOriginP ) originRef;
            if( CdbStatus::Success != originCollRef->find( theOriginName,
                                                           originRef )) {
                cout << errorStr << endl
                     << "    Failed to obtain the origin information for the registry." << endl
                     << "    The Condition/DB might be inproperly initialized or broken." << endl;
                result = CdbStatus::Error;
                break;
            }
            std::string rangeName;
            {
                ooString(32) dbidrange = originRef->dbidrange( );
                assert( 0 != dbidrange.head( ));
                rangeName = dbidrange.head( );
            }

          // Try to locate the registry container

            std::string databaseName = databaseCoreName + rangeName;

            BdbHandle(BdbContObj) cH;

            if( CdbStatus::Success != CdbBdbSUtils::findContainer( cH,
                                                                   databaseName.c_str( ),
                                                                   systemContainerName,
                                                                   theOpenMode )) {
                break;
            }

          // Find the desired object

            BdbHandle(CdbBdbSRegistryP) localRegistryH;

            BdbStatus status = localRegistryH.lookupObj( cH,
                                                         systemRegistryName,
                                                         theOpenMode );
            if(( BdbcSuccess != status ) || BdbIsNull(localRegistryH)) {
                break;
            }
            theRegistryH = localRegistryH;
        }

      // Done

        result = CdbStatus::Success;

    } while( false );

    return result;
}

/////////////////
// End Of File //
/////////////////

---