Improved observers, refactored topic manager and initialization Added

    icegridadmin commands to examine the registries

1 files changed, 267 insertions, 240 deletions

diff --git a/cpp/src/IceGrid/Database.cpp b/cpp/src/IceGrid/Database.cpp
index 77b74d51b5d..d0f64b6e7b8 100644
--- a/cpp/src/IceGrid/Database.cpp
+++ b/cpp/src/IceGrid/Database.cpp
@@ -134,14 +134,14 @@ Database::Database(const Ice::ObjectAdapterPtr& registryAdapter,
     _objectCache(_communicator),
     _allocatableObjectCache(_communicator),
     _serverCache(_communicator, _nodeCache, _adapterCache, _objectCache, _allocatableObjectCache, sessionTimeout),
-    _clientProxy(_communicator->stringToProxy("dummy")),
-    _serverProxy(_communicator->stringToProxy("dummy")),
     _connection(Freeze::createConnection(registryAdapter->getCommunicator(), _envName)),
     _applications(_connection, _applicationDbName),
     _objects(_connection, _objectDbName),
     _adapters(_connection, _adapterDbName),
     _lock(0), 
-    _serial(-1)
+    _applicationSerial(0),
+    _adapterSerial(0),
+    _objectSerial(0)
 {
     ServerEntrySeq entries;
     for(StringApplicationInfoDict::const_iterator p = _applications.begin(); p != _applications.end(); ++p)
@@ -164,6 +164,12 @@ Database::Database(const Ice::ObjectAdapterPtr& registryAdapter,
     _objectCache.setTraceLevels(_traceLevels);
     _allocatableObjectCache.setTraceLevels(_traceLevels);
 
+    _nodeObserverTopic = new NodeObserverTopic(_topicManager, _internalAdapter);
+    _registryObserverTopic = new RegistryObserverTopic(_topicManager);
+    _applicationObserverTopic = new ApplicationObserverTopic(_topicManager, _applications);
+    _adapterObserverTopic = new AdapterObserverTopic(_topicManager, _adapters);
+    _objectObserverTopic = new ObjectObserverTopic(_topicManager, _objects);
+
     //
     // Register a default servant to manage manually registered object adapters.
     //
@@ -197,26 +203,36 @@ Database::getInstanceName() const
     return _instanceName;
 }
 
-RegistryObserverTopicPtr
-Database::getRegistryObserverTopic() const
-{
-    Lock sync(*this);
-    return _registryObserverTopic;
-}
-
-NodeObserverTopicPtr
-Database::getNodeObserverTopic() const
-{
-    Lock sync(*this);
-    return _nodeObserverTopic;
-}
-
 void
-Database::clearTopics()
-{
-    Lock sync(*this);
-    _registryObserverTopic = 0;
-    _nodeObserverTopic = 0;
+Database::destroyTopics()
+{
+    _registryObserverTopic->destroy();
+    _nodeObserverTopic->destroy();
+    _applicationObserverTopic->destroy();
+    _adapterObserverTopic->destroy();
+    _objectObserverTopic->destroy();
+}
+
+ObserverTopicPtr
+Database::getObserverTopic(TopicName name) const
+{
+    switch(name)
+    {
+    case RegistryObserverTopicName:
+	return _registryObserverTopic;
+    case NodeObserverTopicName:
+	return _nodeObserverTopic;
+    case ApplicationObserverTopicName:
+	return _applicationObserverTopic;
+    case AdapterObserverTopicName:
+	return _adapterObserverTopic;
+    case ObjectObserverTopicName:
+	return _objectObserverTopic;
+    default:
+	assert(false);
+	break;
+    }
+    return 0; // Keep the compiler happy.
 }
 
 void
@@ -246,7 +262,7 @@ Database::lock(AdminSessionI* session, const string& userId)
     _lock = session;
     _lockUserId = userId;
     
-    return _serial;
+    return _applicationSerial;
 }
 
 void
@@ -263,138 +279,99 @@ Database::unlock(AdminSessionI* session)
 }
 
 void
-Database::initMaster()
+Database::syncApplications(const ApplicationInfoSeq& applications)
 {
-    Lock sync(*this);
-
-    _nodeObserverTopic = new NodeObserverTopic(_internalAdapter, _topicManager);
-    _registryObserverTopic = new RegistryObserverTopic(_internalAdapter, _topicManager);
-    _serial = 0;
-
-    ApplicationInfoSeq applications;
-    for(StringApplicationInfoDict::const_iterator p = _applications.begin(); p != _applications.end(); ++p)
-    {
-	applications.push_back(p->second);
-    }	
-    AdapterInfoSeq adapters;
-    for(StringAdapterInfoDict::const_iterator q = _adapters.begin(); q != _adapters.end(); ++q)
-    {
-	adapters.push_back(q->second);
-    }
-    ObjectInfoSeq objects;
-    for(IdentityObjectInfoDict::const_iterator r = _objects.begin(); r != _objects.end(); ++r)
+    int serial;
     {
-	objects.push_back(r->second);
-    }
-    _registryObserverTopic->getPublisher()->init(_serial, applications, adapters, objects);
-}
-
-void
-Database::initReplica(int masterSerial, 
-		      const ApplicationInfoSeq& applications,
-		      const AdapterInfoSeq& adapters, 
-		      const ObjectInfoSeq& objects)
-{
-    Lock sync(*this);
-
-    _serial = masterSerial;
+	Lock sync(*this);
 
-    ServerEntrySeq entries;
-    set<string> names;
-    for(ApplicationInfoSeq::const_iterator p = applications.begin(); p != applications.end(); ++p)
-    {
-	try
+	ServerEntrySeq entries;
+	set<string> names;
+	for(ApplicationInfoSeq::const_iterator p = applications.begin(); p != applications.end(); ++p)
 	{
-	    StringApplicationInfoDict::const_iterator s = _applications.find(p->descriptor.name);
-	    if(s != _applications.end())
+	    try
 	    {
-		ApplicationHelper previous(_communicator, s->second.descriptor);
-		ApplicationHelper helper(_communicator, p->descriptor);
-		reload(previous, helper, entries, p->uuid, p->revision);
+		StringApplicationInfoDict::const_iterator s = _applications.find(p->descriptor.name);
+		if(s != _applications.end())
+		{
+		    ApplicationHelper previous(_communicator, s->second.descriptor);
+		    ApplicationHelper helper(_communicator, p->descriptor);
+		    reload(previous, helper, entries, p->uuid, p->revision);
+		}
+		else
+		{
+		    load(ApplicationHelper(_communicator, p->descriptor), entries, p->uuid, p->revision);
+		}
 	    }
-	    else
+	    catch(const DeploymentException& ex)
 	    {
-		load(ApplicationHelper(_communicator, p->descriptor), entries, p->uuid, p->revision);
+		Ice::Warning warn(_traceLevels->logger);
+		warn << "invalid application `" << p->descriptor.name << "':\n" << ex.reason;
 	    }
+	    _applications.put(StringApplicationInfoDict::value_type(p->descriptor.name, *p));
+	    names.insert(p->descriptor.name);
 	}
-	catch(const DeploymentException& ex)
+	StringApplicationInfoDict::iterator s = _applications.begin();
+	while(s != _applications.end())
 	{
-	    Ice::Warning warn(_traceLevels->logger);
-	    warn << "invalid application `" << p->descriptor.name << "':\n" << ex.reason;
+	    if(names.find(s->first) == names.end())
+	    {
+		unload(ApplicationHelper(_communicator, s->second.descriptor), entries);
+		_applications.erase(s++);
+	    }
+	    else
+	    {
+		++s;
+	    }
 	}
-	_applications.put(StringApplicationInfoDict::value_type(p->descriptor.name, *p));
-	names.insert(p->descriptor.name);
+	serial = ++_applicationSerial;
     }
-    StringApplicationInfoDict::iterator s = _applications.begin();
-    while(s != _applications.end())
+
+    _applicationObserverTopic->applicationInit(serial, applications);
+}
+
+void
+Database::syncAdapters(const AdapterInfoSeq& adapters)
+{
+    int serial;
     {
-	if(names.find(s->first) == names.end())
-	{
-	    unload(ApplicationHelper(_communicator, s->second.descriptor), entries);
-	    _applications.erase(s++);
-	}
-	else
+	Lock sync(*this);
+	_adapters.clear();
+	for(AdapterInfoSeq::const_iterator r = adapters.end(); r != adapters.end(); ++r)
 	{
-	    ++s;
+	    _adapters.put(StringAdapterInfoDict::value_type(r->id, *r));
 	}
+	serial = ++_adapterSerial;
     }
     
-    _objects.clear();
-    for(ObjectInfoSeq::const_iterator q = objects.begin(); q != objects.end(); ++q)
-    {
-	_objects.put(IdentityObjectInfoDict::value_type(q->proxy->ice_getIdentity(), *q));
-    }
-    
-    _adapters.clear();
-    for(AdapterInfoSeq::const_iterator r = adapters.end(); r != adapters.end(); ++r)
-    {
-	_adapters.put(StringAdapterInfoDict::value_type(r->id, *r));
-    }
-}
 
-void
-Database::setClientProxy(const Ice::ObjectPrx& proxy)
-{
-    Lock sync(*this);
-    _clientProxy = proxy;
+    _adapterObserverTopic->adapterInit(serial, adapters);
 }
 
 void
-Database::setServerProxy(const Ice::ObjectPrx& proxy)
+Database::syncObjects(const ObjectInfoSeq& objects)
 {
-    Lock sync(*this);
-    _serverProxy = proxy;
-}
-
-Ice::ObjectPrx
-Database::getClientProxy() const
-{
-    Lock sync(*this);
-    return _clientProxy;
+    int serial;
+    {
+	Lock sync(*this);
+	_objects.clear();
+	for(ObjectInfoSeq::const_iterator q = objects.begin(); q != objects.end(); ++q)
+	{
+	    _objects.put(IdentityObjectInfoDict::value_type(q->proxy->ice_getIdentity(), *q));
+	}
+	serial = ++_objectSerial;
+    }
+    _objectObserverTopic->objectInit(serial, objects);
 }
 
 Ice::ObjectPrx
-Database::getServerProxy() const
-{
-    Lock sync(*this);
-    return _serverProxy;
-}
-
-void
-Database::updateReplicatedWellKnownObjects()
+Database::getReplicatedEndpoints(const string& name, const Ice::ObjectPrx& proxy)
 {
-    Ice::ObjectPrx clientProxy = _replicaCache.getClientProxy(getClientProxy());
-    Ice::Identity id;
-    id.category = _instanceName;
-    id.name = "Query";
-    ObjectInfo info;
-    info.type = Query::ice_staticId();
-    info.proxy = clientProxy->ice_identity(id);
-    addObject(info, true);
+    return _replicaCache.getEndpoints(name, proxy);
 }
 
 void
-Database::addApplicationDescriptor(AdminSessionI* session, const ApplicationDescriptor& desc, int masterSerial)
+Database::addApplicationDescriptor(AdminSessionI* session, const ApplicationDescriptor& desc)
 {
     ServerEntrySeq entries;
     string uuid = IceUtil::generateUUID();
@@ -462,7 +439,7 @@ Database::addApplicationDescriptor(AdminSessionI* session, const ApplicationDesc
 
 	_applications.put(StringApplicationInfoDict::value_type(desc.name, info));	
 
-	serial = ++_serial;	
+	serial = ++_applicationSerial;	
 	_updating.erase(desc.name);
 	notifyAll();
     }
@@ -470,10 +447,7 @@ Database::addApplicationDescriptor(AdminSessionI* session, const ApplicationDesc
     //
     // Notify the observers.
     //
-    if(_registryObserverTopic)
-    {
-	_registryObserverTopic->getPublisher()->applicationAdded(serial, info);
-    }
+    _applicationObserverTopic->applicationAdded(serial, info);
 
     if(_traceLevels->application > 0)
     {
@@ -483,9 +457,7 @@ Database::addApplicationDescriptor(AdminSessionI* session, const ApplicationDesc
 }
 
 void
-Database::updateApplicationDescriptor(AdminSessionI* session, 
-				      const ApplicationUpdateDescriptor& update, 
-				      int masterSerial)
+Database::updateApplicationDescriptor(AdminSessionI* session, const ApplicationUpdateDescriptor& update)
 {
     ServerEntrySeq entries;
     ApplicationInfo oldApp;
@@ -597,7 +569,7 @@ Database::instantiateServer(AdminSessionI* session,
 }
 
 void
-Database::removeApplicationDescriptor(AdminSessionI* session, const std::string& name, int masterSerial)
+Database::removeApplicationDescriptor(AdminSessionI* session, const std::string& name)
 {
     ServerEntrySeq entries;
     int serial;
@@ -632,16 +604,13 @@ Database::removeApplicationDescriptor(AdminSessionI* session, const std::string&
 	
 	_applications.erase(p);
 
-	serial = ++_serial;
+	serial = ++_applicationSerial;
     }
 
     //
     // Notify the observers
     //
-    if(_registryObserverTopic)
-    {
-	_registryObserverTopic->getPublisher()->applicationRemoved(serial, name);
-    }
+    _applicationObserverTopic->applicationRemoved(serial, name);
 
     if(_traceLevels->application > 0)
     {
@@ -680,10 +649,18 @@ Database::addNode(const string& name, const NodeSessionIPtr& session)
 {
     _nodeCache.get(name, true)->setSession(session);
 
-    ObjectInfo info;
-    info.type = Node::ice_staticId();
-    info.proxy = session->getNode();
-    addObject(info, true);
+//     //
+//     // Only the master adds the node well-known proxy to its
+//     // database. The well-known proxy will be transmitted to the
+//     // replicas through the replication of the database.
+//     //
+//     if(_master)
+//     {
+	ObjectInfo info;
+	info.type = Node::ice_staticId();
+	info.proxy = session->getNode();
+	addObject(info, true);
+//    }
 }
 
 NodePrx 
@@ -716,65 +693,53 @@ Database::removeNode(const string& name, const NodeSessionIPtr& session, bool sh
     // observer to ensure that only nodes which are up are teared
     // down).
     //
-    NodeObserverTopicPtr topic = getNodeObserverTopic();
-    if(topic)
-    {
-	topic->getPublisher()->nodeDown(name);
-    }
+    _nodeObserverTopic->nodeDown(name);
 
     _nodeCache.get(name)->setSession(0);
 }
 
+Ice::StringSeq
+Database::getAllNodes(const string& expression)
+{
+    return _nodeCache.getAll(expression);
+}
+
 void
 Database::addReplica(const string& name, const ReplicaSessionIPtr& session)
 {
     _replicaCache.add(name, session);
 
-    //
-    // Only the master adds the node well-known proxy to its
-    // database. The well-known proxy will be transmitted to the
-    // replicas through the replication of the database.
-    //
-    if(_master)
-    {
-	ObjectInfo info;
-	info.type = InternalRegistry::ice_staticId();
-	info.proxy = session->getProxy();
-	addObject(info, true);
-    }
+    _applicationObserverTopic->subscribe(session->getObserver());
+    _adapterObserverTopic->subscribe(session->getObserver());
+    _objectObserverTopic->subscribe(session->getObserver());
+}
 
-    RegistryObserverTopicPtr topic = getRegistryObserverTopic();
-    if(topic)
-    {
-	topic->subscribe(session->getObserver());
-    }
+InternalRegistryPrx 
+Database::getReplica(const string& name) const
+{
+    return _replicaCache.get(name)->getProxy();
 }
 
-void
-Database::removeReplica(const string& name, const ReplicaSessionIPtr& session, bool shutdown)
+RegistryInfo
+Database::getReplicaInfo(const string& name) const
 {
-    //
-    // If this registry isn't being shutdown remove the replica
-    // well-known proxy from the database.
-    //
-    if(!shutdown)
-    {
-	removeObject(session->getProxy()->ice_getIdentity());
-    }
+    return _replicaCache.get(name)->getInfo();
+}
 
-    RegistryObserverTopicPtr topic = getRegistryObserverTopic();
-    if(topic)
-    {
-	topic->unsubscribe(session->getObserver());
-    }
+void
+Database::removeReplica(const string& name, const ReplicaSessionIPtr& session)
+{
+    _applicationObserverTopic->unsubscribe(session->getObserver());
+    _adapterObserverTopic->unsubscribe(session->getObserver());
+    _objectObserverTopic->unsubscribe(session->getObserver());
 
     _replicaCache.remove(name);
 }
 
-Ice::StringSeq 
-Database::getAllNodes(const string& expression)
+Ice::StringSeq
+Database::getAllReplicas(const string& expression)
 {
-    return _nodeCache.getAll(expression);
+    return _replicaCache.getAll(expression);
 }
 
 ServerInfo
@@ -810,8 +775,7 @@ Database::getAllNodeServers(const string& node)
 }
 
 bool
-Database::setAdapterDirectProxy(const string& adapterId, const string& replicaGroupId, const Ice::ObjectPrx& proxy, 
-				int masterSerial)
+Database::setAdapterDirectProxy(const string& adapterId, const string& replicaGroupId, const Ice::ObjectPrx& proxy)
 {
     AdapterInfo info;
     int serial;
@@ -851,7 +815,7 @@ Database::setAdapterDirectProxy(const string& adapterId, const string& replicaGr
 	    _adapters.erase(p);
 	}
 
-	serial = ++_serial;
+	serial = ++_adapterSerial;
     }
 
     if(_traceLevels->adapter > 0)
@@ -864,24 +828,21 @@ Database::setAdapterDirectProxy(const string& adapterId, const string& replicaGr
 	}
     }
 
-    if(_registryObserverTopic)
+    if(proxy)
     {
-	if(proxy)
+	if(updated)
 	{
-	    if(updated)
-	    {
-		_registryObserverTopic->getPublisher()->adapterUpdated(serial, info);
-	    }
-	    else
-	    {
-		_registryObserverTopic->getPublisher()->adapterAdded(serial, info);
-	    }
+	    _adapterObserverTopic->adapterUpdated(serial, info);
 	}
 	else
 	{
-	    _registryObserverTopic->getPublisher()->adapterRemoved(serial, adapterId);
+	    _adapterObserverTopic->adapterAdded(serial, info);
 	}
     }
+    else
+    {
+	_adapterObserverTopic->adapterRemoved(serial, adapterId);
+    }
     return true;
 }
 
@@ -943,12 +904,12 @@ Database::removeAdapter(const string& adapterId)
 
 	if(infos.empty())
 	{
-	    serial = ++_serial;
+	    serial = ++_adapterSerial;
 	}
 	else
 	{
-	    serial = _serial;
-	    _serial += static_cast<int>(static_cast<int>(infos.size()));
+	    serial = _adapterSerial;
+	    _adapterSerial += static_cast<int>(static_cast<int>(infos.size()));
 	}
     }
 
@@ -958,18 +919,15 @@ Database::removeAdapter(const string& adapterId)
 	out << "removed " << (infos.empty() ? "adapter" : "replica group") << " `" << adapterId << "'";
     }
 
-    if(_registryObserverTopic)
+    if(infos.empty())
     {
-	if(infos.empty())
-	{
-	    _registryObserverTopic->getPublisher()->adapterRemoved(serial, adapterId);
-	}
-	else
+	_adapterObserverTopic->adapterRemoved(serial, adapterId);
+    }
+    else
+    {
+	for(AdapterInfoSeq::const_iterator p = infos.begin(); p != infos.end(); ++p)
 	{
-	    for(AdapterInfoSeq::const_iterator p = infos.begin(); p != infos.end(); ++p)
-	    {
-		_registryObserverTopic->getPublisher()->adapterUpdated(++serial, *p);
-	    }
+	    _adapterObserverTopic->adapterUpdated(++serial, *p);
 	}
     }
 }
@@ -1125,7 +1083,7 @@ Database::getAllAdapters(const string& expression)
 }
 
 void
-Database::addObject(const ObjectInfo& info, bool replaceIfExistsInDatabase, int masterSerial)
+Database::addObject(const ObjectInfo& info, bool replaceIfExistsInDatabase)
 {
     int serial;
     const Ice::Identity id = info.proxy->ice_getIdentity();
@@ -1150,22 +1108,19 @@ Database::addObject(const ObjectInfo& info, bool replaceIfExistsInDatabase, int
 	}
 	_objects.put(IdentityObjectInfoDict::value_type(id, info));
 	
-	serial = ++_serial;
+	serial = ++_objectSerial;
     }
 	
     //
     // Notify the observers.
     //
-    if(_registryObserverTopic)
+    if(!update)
     {
-	if(!update)
-	{
-	    _registryObserverTopic->getPublisher()->objectAdded(serial, info);
-	}
-	else
-	{
-	    _registryObserverTopic->getPublisher()->objectUpdated(serial, info);
-	}
+	_objectObserverTopic->objectAdded(serial, info);
+    }
+    else
+    {
+	_objectObserverTopic->objectUpdated(serial, info);
     }
 
     if(_traceLevels->object > 0)
@@ -1183,7 +1138,7 @@ Database::addObject(const ObjectInfo& info, bool replaceIfExistsInDatabase, int
 }
 
 void
-Database::removeObject(const Ice::Identity& id, int masterSerial)
+Database::removeObject(const Ice::Identity& id)
 {
     int serial;
     {
@@ -1207,16 +1162,13 @@ Database::removeObject(const Ice::Identity& id, int masterSerial)
 	}
 	_objects.erase(p);
 
-	serial = ++_serial;
+	serial = ++_objectSerial;
     }
 
     //
     // Notify the observers.
     //
-    if(_registryObserverTopic)
-    {
-	_registryObserverTopic->getPublisher()->objectRemoved(serial, id);
-    }
+    _objectObserverTopic->objectRemoved(serial, id);
 
     if(_traceLevels->object > 0)
     {
@@ -1226,7 +1178,7 @@ Database::removeObject(const Ice::Identity& id, int masterSerial)
 }
 
 void
-Database::updateObject(const Ice::ObjectPrx& proxy, int masterSerial)
+Database::updateObject(const Ice::ObjectPrx& proxy)
 {
     const Ice::Identity id = proxy->ice_getIdentity();
     int serial;
@@ -1255,16 +1207,13 @@ Database::updateObject(const Ice::ObjectPrx& proxy, int masterSerial)
 	info.proxy = proxy;
 	p.set(info);
 
-	serial = ++_serial;
+	serial = ++_objectSerial;
     }
 
     //
     // Notify the observers.
     //
-    if(_registryObserverTopic)
-    {
-	_registryObserverTopic->getPublisher()->objectUpdated(serial, info);
-    }
+    _objectObserverTopic->objectUpdated(serial, info);
 
     if(_traceLevels->object > 0)
     {
@@ -1274,6 +1223,88 @@ Database::updateObject(const Ice::ObjectPrx& proxy, int masterSerial)
 }
 
 void
+Database::addOrUpdateObjectsInDatabase(const ObjectInfoSeq& objects)
+{
+    int serial;
+    vector<bool> updated;
+    {
+	Lock sync(*this);
+
+	Freeze::TransactionHolder txHolder(_connection);
+	for(ObjectInfoSeq::const_iterator p = objects.begin(); p != objects.end(); ++p)
+	{
+	    updated.push_back(_objects.find(p->proxy->ice_getIdentity()) != _objects.end());
+	    _objects.put(IdentityObjectInfoDict::value_type(p->proxy->ice_getIdentity(), *p));
+	}
+	serial = _objectSerial;
+	_objectSerial += static_cast<int>(static_cast<int>(objects.size()));
+	txHolder.commit();
+    }
+	
+    //
+    // Notify the observers.
+    //
+    vector<bool>::const_iterator q = updated.begin();
+    for(ObjectInfoSeq::const_iterator p = objects.begin(); p != objects.end(); ++p, ++q)
+    {
+	//
+	// TODO: Add a better observer call?
+	//
+	if(!*q)
+	{
+	    _objectObserverTopic->objectAdded(++serial, *p);
+	}
+	else
+	{
+	    _objectObserverTopic->objectUpdated(++serial, *p);
+	}
+
+	if(_traceLevels->object > 0)
+	{
+	    Ice::Trace out(_traceLevels->logger, _traceLevels->objectCat);
+	    if(!*q)
+	    {
+		out << "added object `" << _communicator->identityToString(p->proxy->ice_getIdentity()) << "'";
+	    }
+	    else
+	    {
+		out << "updated object `" << _communicator->identityToString(p->proxy->ice_getIdentity()) << "'";
+	    }
+	}
+    }
+}
+
+void
+Database::removeObjectsInDatabase(const ObjectInfoSeq& objects)
+{
+    int serial;
+    {
+	Freeze::TransactionHolder txHolder(_connection);
+	for(ObjectInfoSeq::const_iterator p = objects.begin(); p != objects.end(); ++p)
+	{
+	    _objects.erase(p->proxy->ice_getIdentity());
+	}
+	serial = _objectSerial;
+	_objectSerial += static_cast<int>(static_cast<int>(objects.size()));
+	txHolder.commit();
+    }
+	
+    //
+    // Notify the observers.
+    //
+    for(ObjectInfoSeq::const_iterator p = objects.begin(); p != objects.end(); ++p)
+    {
+	_objectObserverTopic->objectRemoved(++serial, p->proxy->ice_getIdentity());
+
+	if(_traceLevels->object > 0)
+	{
+	    Ice::Trace out(_traceLevels->logger, _traceLevels->objectCat);
+	    out << "removed object `" << _communicator->identityToString(p->proxy->ice_getIdentity()) << "'";
+	}
+    }
+}
+
+void
 Database::allocateObject(const Ice::Identity& id, const ObjectAllocationRequestPtr& request)
 {
     _allocatableObjectCache.get(id)->allocate(request);
@@ -1741,7 +1772,7 @@ Database::finishUpdate(ServerEntrySeq& entries,
 	updateInfo.descriptor = update;
 
 	_applications.put(StringApplicationInfoDict::value_type(update.name, info));
-	serial = ++_serial;
+	serial = ++_applicationSerial;
 	_updating.erase(update.name);
 	notifyAll();
     }
@@ -1749,10 +1780,7 @@ Database::finishUpdate(ServerEntrySeq& entries,
     //
     // Notify the observers.
     //
-    if(_registryObserverTopic)
-    {
-	_registryObserverTopic->getPublisher()->applicationUpdated(serial, updateInfo);
-    }
+    _applicationObserverTopic->applicationUpdated(serial, updateInfo);
 
     if(_traceLevels->application > 0)
     {
@@ -1760,4 +1788,3 @@ Database::finishUpdate(ServerEntrySeq& entries,
 	out << "updated application `" << update.name << "'";
     }
 }
-