cache.cxx

来自「vovida的软交换」· CXX 代码 · 共 591 行 · 第 1/2 页

	Record::stringRecordType(recType).c_str(), 
	hostname.c_str());

  string spid = itos(getpid());
  
  // Build the unique key for this listener ... ( Hostname + RecordType ) 
  KeyType key = spid + "@" + hostname + ":" + Record::stringRecordType(recType);
   
  // Check if we already have an entry in the database or our local list
  // Either it'll be in the our local listener list or in the 
  // database. NOTE: UpdateRegInfoRec's are not cached the same as 
  // other records.
  if ((listeners.find(key) != listeners.end()) ||
      (getRecord(UPDATE_REG_INFO_RECORD, key, false) != NULL)) { 
    // Already a listener active for this, return false
    // Could amend the existing listener ?
    cpLog(LOG_DEBUG, "Register failed, listener already active");
    return false;
  }

  // start a CacheListener to listen for this update
  CacheListener* cl = new CacheListener(myCache);

  // It'll find a port num
  int portNum = cl->getListenPort();
    
  if (portNum == 0) { 
    // No ports open 
    delete cl;
    return false;
  }

  // Create a record for storing the information
  UpdateRegInfoRec* updateRec = new UpdateRegInfoRec();
  
  // update this record with the information supplied
  updateRec->setKey(key);
  updateRec->setCallBackFunc(callBackFunc);
  updateRec->setHostname(hostname);
  updateRec->setListenerPort(portNum);
  updateRec->setTriggerRecordType(recType);
  cl->setUpdateRecord(updateRec);

  // Start the listener ...
  cl->run();

  // Now send that record to the database so the database triggers can find 
  // it.
  if (!(insertRecord(updateRec))) {
    cpLog(LOG_DEBUG, 
	  "Register failed as database insert failed, stopping listener thread.");
    // Failed, stop the listener and return
    delete updateRec;
    cl->shutdown(); cl->join();
    delete cl;
    return false;
  }

  // We've added the record to the database, now add a copy of it to our local 
  // store.
  listeners.insert(listenersList::value_type(key, cl));

  delete updateRec;
  return true;
}

bool Cache::deregisterForUpdates(string hostname, RecordType recType) { 

  string spid = itos(getpid()); 

  // Build the unique key for this listener ... ( Hostname + RecordType ) 
  KeyType key = spid + "@" + hostname + ":" + Record::stringRecordType(recType);

  cpLog(LOG_DEBUG, 
	"Deregistering for update messages for records of type %s on %s",
	Record::stringRecordType(recType).c_str(), 
	hostname.c_str());

  // Delete the record from the database
  UpdateRegInfoRec* updateRec = new UpdateRegInfoRec;
  updateRec->setKey(key);
  DBConnection* dbconn = myDBManager->searchDBConPool(dbinfo);

  if (!dbconn) {
    delete updateRec;
    cpLog(LOG_ERR, "Can't get database connection");
    return false;
  }

  if (!( updateRec->deleteRecord(dbconn)) ) {     
    // Failed 
    cpLog(LOG_DEBUG, "Database delete failed, listener not deregistered");
    myDBManager->returnDBCon(dbconn);
    delete updateRec;
    return false;
  }
  delete updateRec;
  myDBManager->returnDBCon(dbconn);

  // Try and get the listener from our local listeners list
  listenersList::iterator i = listeners.find(key);

  if (i == listeners.end()) {
    cpLog(LOG_ERR, "Can't find listener in local list, can't shutdown listener thread");
    return false;
  }
  
  CacheListener* cl = i->second;
  // Delete the listener, it'll shutdown automatically
  delete cl;
  listeners.erase(key);

  return true;
}

void Cache::stopAllListeners() { 
  listenersList::iterator i = listeners.begin();

  cpLog(LOG_DEBUG, "Stopping all listener threads");

  // Stop all the listeners and delete the 
  for (;i != listeners.end();i++) { 

    // Remove from the database
    DBConnection* dbconn = myDBManager->searchDBConPool(dbinfo);
    if (dbconn) { 
      UpdateRegInfoRec* updateRec = new UpdateRegInfoRec;
      updateRec->setKey(i->first);
      if (!(updateRec->deleteRecord(dbconn))) {
	cpLog(LOG_DEBUG, "Can't delete database entry for listener, ignoring.");
      };
      myDBManager->returnDBCon(dbconn);
      
      // Shutdown the listener and delete the pointers
      delete(i->second); delete updateRec;

    } else { 
      cpLog(LOG_ERR, "Can't get database connection");
    }

  }

  /****
   * We can't just delete the updates table in the database because
   * there may be other active cache processes using it.
   * TODO: We should delete all entries where the process ID matches
   * this process. 
   * The table is cleared on vocal start up anyway which should stop 
   * it filling up if the system fails.
   ****/
//   DBConnection* dbconn = myDBManager->searchDBConPool(dbinfo);
//   if (dbconn) { 
//     dbconn->DBSQLDelete((string) "DELETE FROM updatereg *");
//     myDBManager->returnDBCon(dbconn);
//   } else { 
//     cpLog(LOG_ERR, "Can't get database connection");
//   }

  listeners.empty();
}

bool Cache::initCache(RecordType recType) { 

  Record* rec = Record::createNewRecord(recType); // Get a blank record

  cpLog(LOG_DEBUG, "Initalizing cache for records of type %s", 
	Record::stringRecordType(recType).c_str());

  DBConnection * dbconn = myDBManager->searchDBConPool(dbinfo);
  
  if (!dbconn) { 
    cpLog(LOG_ERR, "Can't get database connection");
    return false;
  }

  KeyList keyList = rec->getAllKeys(dbconn);
  myDBManager->returnDBCon(dbconn);

  KeyList::iterator i = keyList.begin();

  /*
   * This for loop retrieves each record in turn from the database and 
   * creates the record object. These will automatically be cached, the 
   * created object is disgarded.
   */
  // Dispose of the temporary object
  delete rec; 

  for (;i != keyList.end(); i++) {
    rec = getRecord(recType, *i);
    delete rec;
  }

  return true;
}

KeyList Cache::getAllKeys(RecordType recType)
{

  Record* rec = Record::createNewRecord(recType); // Get a blank record
  cpLog(LOG_DEBUG, "Getting all keys for record of type %s", 
        Record::stringRecordType(recType).c_str());
        
  DBConnection * dbconn = myDBManager->searchDBConPool(dbinfo);
  if (!dbconn) { 
    KeyList keyList;
    cpLog(LOG_ERR, "Can't get database connection");
    return keyList;
  }
  KeyList keyList = rec->getAllKeys(dbconn);
  myDBManager->returnDBCon(dbconn);

  return keyList;
}
/**
 * TODO: Currently this gets all the information from the database, 
 *       however if we make sure that the triggers inform us of alias
 *       INSERTS as well as UPDATE/DELETE then we could make this work
 **/
KeyList Cache::getAliasesForUser(string user) { 

  DBConnection* dbconn = myDBManager->searchDBConPool(dbinfo);
  KeyList aliases;
  DBResultset* results; 

  // If we get a valid connection
  if (dbconn) { 
    try { 
      results = dbconn->DBSQLSelect((string)" "                                 + 
				    "SELECT l.alias FROM ALIASES l, ACCOUNTS a" + 
				    " WHERE a.id = l.account_id AND "           +
				    "       a.username = '"+user+"' "           +
				    ";"
				    );
    }
    catch ( VException &e ) { 
      cpLog(LOG_ERR, "DBSQLSelect (for getAliasesForUser('%s')) failed due to %s",user.c_str(),e.getDescription().c_str());
      myDBManager->returnDBCon(dbconn);
      return aliases;
    }

    ResultTabletype recs = results->GetRecords();
    delete results;  // Delete the results object in case this fails

    for (ResultTabletype::iterator i = recs.begin();i != recs.end();i++) { 
      // i should point to a single string in a list, check this
      aliases.push_back(*((*i).begin()));
    }
    myDBManager->returnDBCon(dbconn);     
  }
  return aliases;
}

/*
 * Constructor / Destructor 
 */
    
Cache::Cache() { 
  /*
  Cache::Cache("PGSQL", "vocal", "localhost", "vocal", "vocal");
}

Cache::Cache(string dbType,
	     string dbName, 
	     string dbHost, 
	     string dbUser, 
	     string dbPasswd) { 
  */
  char hostname[100] = "127.0.0.1";
  if ((gethostname(hostname, 100)) < 0) { 
    cpLog(LOG_DEBUG, "Can't get hostname, using 127.0.0.1");
  }
  cpLog(LOG_DEBUG, "Constructing Cache");
  // Reset the data for the singleton pattern

  myCache = new CacheStorage;
  myDBManager = ConnectionManager::getInstance();
  //  myDBThread = new DBConnThread(myDBManager);

  myCacheManager = new CacheManager(myCache);

  // Init the database information
  dbinfo.dbBackend = "PGSQL"; //= dbType;
  dbinfo.database = "vocal"; //= dbName;
  dbinfo.host = hostname; //= dbHost;
  dbinfo.user = "vocal"; //=dbUser;
  dbinfo.password = "vocal"; //= dbPasswd;

  // Start the seperate manager threads running
  myCacheManager->run();
  //  myDBThread->run();
}
Cache::~Cache() { 
}