summaryrefslogtreecommitdiff
path: root/cpp/src
diff options
context:
space:
mode:
Diffstat (limited to 'cpp/src')
-rw-r--r--cpp/src/IceGrid/AdminSessionI.cpp14
-rw-r--r--cpp/src/IceGrid/AdminSessionI.h2
-rw-r--r--cpp/src/IceGrid/Topics.cpp230
-rw-r--r--cpp/src/IceGrid/Topics.h24
4 files changed, 187 insertions, 83 deletions
diff --git a/cpp/src/IceGrid/AdminSessionI.cpp b/cpp/src/IceGrid/AdminSessionI.cpp
index 8cabcba8fee..f201d595148 100644
--- a/cpp/src/IceGrid/AdminSessionI.cpp
+++ b/cpp/src/IceGrid/AdminSessionI.cpp
@@ -185,11 +185,11 @@ AdminSessionI::setObserversByIdentity(const Ice::Identity& registryObserver,
throw ex;
}
- setupObserverSubscription(RegistryObserverTopicName, toProxy(registryObserver, current.con));
- setupObserverSubscription(NodeObserverTopicName, toProxy(nodeObserver, current.con));
- setupObserverSubscription(ApplicationObserverTopicName, toProxy(appObserver, current.con));
- setupObserverSubscription(AdapterObserverTopicName, toProxy(adapterObserver, current.con));
- setupObserverSubscription(ObjectObserverTopicName, toProxy(objectObserver, current.con));
+ setupObserverSubscription(RegistryObserverTopicName, toProxy(registryObserver, current.con, current.encoding));
+ setupObserverSubscription(NodeObserverTopicName, toProxy(nodeObserver, current.con, current.encoding));
+ setupObserverSubscription(ApplicationObserverTopicName, toProxy(appObserver, current.con, current.encoding));
+ setupObserverSubscription(AdapterObserverTopicName, toProxy(adapterObserver, current.con, current.encoding));
+ setupObserverSubscription(ObjectObserverTopicName, toProxy(objectObserver, current.con, current.encoding));
}
@@ -335,9 +335,9 @@ AdminSessionI::setupObserverSubscription(TopicName name, const Ice::ObjectPrx& o
}
Ice::ObjectPrx
-AdminSessionI::toProxy(const Ice::Identity& id, const Ice::ConnectionPtr& connection)
+AdminSessionI::toProxy(const Ice::Identity& id, const Ice::ConnectionPtr& connection, const Ice::EncodingVersion& v)
{
- return id.name.empty() ? Ice::ObjectPrx() : connection->createProxy(id);
+ return id.name.empty() ? Ice::ObjectPrx() : connection->createProxy(id)->ice_encodingVersion(v);
}
FileIteratorPrx
diff --git a/cpp/src/IceGrid/AdminSessionI.h b/cpp/src/IceGrid/AdminSessionI.h
index ff234953ab3..bc63acce993 100644
--- a/cpp/src/IceGrid/AdminSessionI.h
+++ b/cpp/src/IceGrid/AdminSessionI.h
@@ -66,7 +66,7 @@ public:
private:
void setupObserverSubscription(TopicName, const Ice::ObjectPrx&);
- Ice::ObjectPrx toProxy(const Ice::Identity&, const Ice::ConnectionPtr&);
+ Ice::ObjectPrx toProxy(const Ice::Identity&, const Ice::ConnectionPtr&, const Ice::EncodingVersion&);
FileIteratorPrx addFileIterator(const FileReaderPrx&, const std::string&, int, const Ice::Current&);
virtual void destroyImpl(bool);
diff --git a/cpp/src/IceGrid/Topics.cpp b/cpp/src/IceGrid/Topics.cpp
index 62c582858ac..d2b976c7452 100644
--- a/cpp/src/IceGrid/Topics.cpp
+++ b/cpp/src/IceGrid/Topics.cpp
@@ -14,28 +14,48 @@
using namespace std;
using namespace IceGrid;
+namespace
+{
+
+//
+// Encodings supported by the observers. We create one topic per
+// encoding version and subscribe the observer to the appropriate
+// topic depending on its encoding.
+//
+Ice::EncodingVersion encodings[] = {
+ { 1, 0 },
+ { 1, 1 }
+};
+
+}
ObserverTopic::ObserverTopic(const IceStorm::TopicManagerPrx& topicManager, const string& name) :
_logger(topicManager->ice_getCommunicator()->getLogger()),
_serial(0)
{
- IceStorm::TopicPrx t;
- try
- {
- t = topicManager->create(name);
- }
- catch(const IceStorm::TopicExists&)
+ for(int i = 0; i < sizeof(encodings) / sizeof(Ice::EncodingVersion); ++i)
{
- t = topicManager->retrieve(name);
- }
+ ostringstream os;
+ os << name << "-" << Ice::encodingVersionToString(encodings[i]);
+ IceStorm::TopicPrx t;
+ try
+ {
+ t = topicManager->create(os.str());
+ }
+ catch(const IceStorm::TopicExists&)
+ {
+ t = topicManager->retrieve(os.str());
+ }
- //
- // NOTE: collocation optimization needs to be turned on for the
- // topic because the subscribe() method is given a fixed proxy
- // which can't be marshalled.
- //
- _topic = IceStorm::TopicPrx::uncheckedCast(t->ice_collocationOptimized(true));
- _basePublisher = _topic->getPublisher()->ice_collocationOptimized(false);
+ //
+ // NOTE: collocation optimization needs to be turned on for the
+ // topic because the subscribe() method is given a fixed proxy
+ // which can't be marshalled.
+ //
+ _topics[encodings[i]] = t->ice_collocationOptimized(true);
+ _basePublishers.push_back(
+ t->getPublisher()->ice_collocationOptimized(false)->ice_encodingVersion(encodings[i]));
+ }
}
ObserverTopic::~ObserverTopic()
@@ -46,7 +66,7 @@ int
ObserverTopic::subscribe(const Ice::ObjectPrx& obsv, const string& name)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return -1;
}
@@ -56,7 +76,15 @@ ObserverTopic::subscribe(const Ice::ObjectPrx& obsv, const string& name)
{
IceStorm::QoS qos;
qos["reliability"] = "ordered";
- initObserver(_topic->subscribeAndGetPublisher(qos, obsv->ice_twoway()));
+ Ice::EncodingVersion v = IceInternal::getCompatibleEncoding(obsv->ice_getEncodingVersion());
+ map<Ice::EncodingVersion, IceStorm::TopicPrx>::const_iterator p = _topics.find(v);
+ if(p == _topics.end())
+ {
+ Ice::Warning out(_logger);
+ out << "unsupported encoding version for observer `" << obsv << "'";
+ return -1;
+ }
+ initObserver(p->second->subscribeAndGetPublisher(qos, obsv->ice_twoway()));
}
catch(const IceStorm::AlreadySubscribed&)
{
@@ -77,10 +105,13 @@ void
ObserverTopic::unsubscribe(const Ice::ObjectPrx& observer, const string& name)
{
Lock sync(*this);
- if(_topic)
+ Ice::EncodingVersion v = IceInternal::getCompatibleEncoding(observer->ice_getEncodingVersion());
+ map<Ice::EncodingVersion, IceStorm::TopicPrx>::const_iterator p = _topics.find(v);
+ if(p == _topics.end())
{
- _topic->unsubscribe(observer);
+ return;
}
+ p->second->unsubscribe(observer);
assert(observer);
@@ -116,7 +147,7 @@ void
ObserverTopic::destroy()
{
Lock sync(*this);
- _topic = 0;
+ _topics.clear();
notifyAll();
}
@@ -186,7 +217,7 @@ ObserverTopic::waitForSyncedSubscribersNoSync(int serial, const string& name)
//
// Wait until all the updates are received or the service shutdown.
//
- while(_topic)
+ while(!_topics.empty())
{
map<int, set<string> >::const_iterator p = _waitForUpdates.find(serial);
if(p == _waitForUpdates.end())
@@ -239,14 +270,14 @@ ObserverTopic::getContext(int serial) const
RegistryObserverTopic::RegistryObserverTopic(const IceStorm::TopicManagerPrx& topicManager) :
ObserverTopic(topicManager, "RegistryObserver")
{
- const_cast<RegistryObserverPrx&>(_publisher) = RegistryObserverPrx::uncheckedCast(_basePublisher);
+ _publishers = getPublishers<RegistryObserverPrx>();
}
void
RegistryObserverTopic::registryUp(const RegistryInfo& info)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return;
}
@@ -254,7 +285,10 @@ RegistryObserverTopic::registryUp(const RegistryInfo& info)
_registries.insert(make_pair(info.name, info));
try
{
- _publisher->registryUp(info);
+ for(vector<RegistryObserverPrx>::const_iterator p = _publishers.begin(); p != _publishers.end(); ++p)
+ {
+ (*p)->registryUp(info);
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -267,7 +301,7 @@ void
RegistryObserverTopic::registryDown(const string& name)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return;
}
@@ -281,7 +315,10 @@ RegistryObserverTopic::registryDown(const string& name)
_registries.erase(name);
try
{
- _publisher->registryDown(name);
+ for(vector<RegistryObserverPrx>::const_iterator p = _publishers.begin(); p != _publishers.end(); ++p)
+ {
+ (*p)->registryDown(name);
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -307,7 +344,7 @@ NodeObserverTopic::NodeObserverTopic(const IceStorm::TopicManagerPrx& topicManag
const Ice::ObjectAdapterPtr& adapter) :
ObserverTopic(topicManager, "NodeObserver")
{
- const_cast<NodeObserverPrx&>(_publisher) = NodeObserverPrx::uncheckedCast(_basePublisher);
+ _publishers = getPublishers<NodeObserverPrx>();
try
{
const_cast<NodeObserverPrx&>(_externalPublisher) = NodeObserverPrx::uncheckedCast(adapter->addWithUUID(this));
@@ -327,7 +364,7 @@ void
NodeObserverTopic::nodeUp(const NodeDynamicInfo& info, const Ice::Current&)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return;
}
@@ -335,7 +372,10 @@ NodeObserverTopic::nodeUp(const NodeDynamicInfo& info, const Ice::Current&)
_nodes.insert(make_pair(info.info.name, info));
try
{
- _publisher->nodeUp(info);
+ for(vector<NodeObserverPrx>::const_iterator p = _publishers.begin(); p != _publishers.end(); ++p)
+ {
+ (*p)->nodeUp(info);
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -354,7 +394,7 @@ void
NodeObserverTopic::updateServer(const string& node, const ServerDynamicInfo& server, const Ice::Current&)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return;
}
@@ -394,7 +434,10 @@ NodeObserverTopic::updateServer(const string& node, const ServerDynamicInfo& ser
try
{
- _publisher->updateServer(node, server);
+ for(vector<NodeObserverPrx>::const_iterator p = _publishers.begin(); p != _publishers.end(); ++p)
+ {
+ (*p)->updateServer(node, server);
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -407,7 +450,7 @@ void
NodeObserverTopic::updateAdapter(const string& node, const AdapterDynamicInfo& adapter, const Ice::Current&)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return;
}
@@ -447,7 +490,10 @@ NodeObserverTopic::updateAdapter(const string& node, const AdapterDynamicInfo& a
try
{
- _publisher->updateAdapter(node, adapter);
+ for(vector<NodeObserverPrx>::const_iterator p = _publishers.begin(); p != _publishers.end(); ++p)
+ {
+ (*p)->updateAdapter(node, adapter);
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -460,7 +506,7 @@ void
NodeObserverTopic::nodeDown(const string& name)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return;
}
@@ -472,7 +518,10 @@ NodeObserverTopic::nodeDown(const string& name)
_nodes.erase(name);
try
{
- _publisher->nodeDown(name);
+ for(vector<NodeObserverPrx>::const_iterator p = _publishers.begin(); p != _publishers.end(); ++p)
+ {
+ (*p)->nodeDown(name);
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -500,14 +549,14 @@ ApplicationObserverTopic::ApplicationObserverTopic(const IceStorm::TopicManagerP
ObserverTopic(topicManager, "ApplicationObserver"),
_applications(applications)
{
- const_cast<ApplicationObserverPrx&>(_publisher) = ApplicationObserverPrx::uncheckedCast(_basePublisher);
+ _publishers = getPublishers<ApplicationObserverPrx>();
}
int
ApplicationObserverTopic::applicationInit(int serial, const ApplicationInfoSeq& apps)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return -1;
}
@@ -519,7 +568,10 @@ ApplicationObserverTopic::applicationInit(int serial, const ApplicationInfoSeq&
}
try
{
- _publisher->applicationInit(serial, apps, getContext(serial));
+ for(vector<ApplicationObserverPrx>::const_iterator p = _publishers.begin(); p != _publishers.end(); ++p)
+ {
+ (*p)->applicationInit(serial, apps, getContext(serial));
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -534,7 +586,7 @@ int
ApplicationObserverTopic::applicationAdded(int serial, const ApplicationInfo& info)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return -1;
}
@@ -543,7 +595,10 @@ ApplicationObserverTopic::applicationAdded(int serial, const ApplicationInfo& in
_applications.insert(make_pair(info.descriptor.name, info));
try
{
- _publisher->applicationAdded(serial, info, getContext(serial));
+ for(vector<ApplicationObserverPrx>::const_iterator p = _publishers.begin(); p != _publishers.end(); ++p)
+ {
+ (*p)->applicationAdded(serial, info, getContext(serial));
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -558,7 +613,7 @@ int
ApplicationObserverTopic::applicationRemoved(int serial, const string& name)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return -1;
}
@@ -566,7 +621,10 @@ ApplicationObserverTopic::applicationRemoved(int serial, const string& name)
_applications.erase(name);
try
{
- _publisher->applicationRemoved(serial, name, getContext(serial));
+ for(vector<ApplicationObserverPrx>::const_iterator p = _publishers.begin(); p != _publishers.end(); ++p)
+ {
+ (*p)->applicationRemoved(serial, name, getContext(serial));
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -581,7 +639,7 @@ int
ApplicationObserverTopic::applicationUpdated(int serial, const ApplicationUpdateInfo& info)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return -1;
}
@@ -592,7 +650,7 @@ ApplicationObserverTopic::applicationUpdated(int serial, const ApplicationUpdate
map<string, ApplicationInfo>::iterator p = _applications.find(info.descriptor.name);
if(p != _applications.end())
{
- ApplicationHelper helper(_publisher->ice_getCommunicator(), p->second.descriptor);
+ ApplicationHelper helper(_publishers[0]->ice_getCommunicator(), p->second.descriptor);
p->second.descriptor = helper.update(info.descriptor);
p->second.updateTime = info.updateTime;
p->second.updateUser = info.updateUser;
@@ -625,7 +683,10 @@ ApplicationObserverTopic::applicationUpdated(int serial, const ApplicationUpdate
}
try
{
- _publisher->applicationUpdated(serial, info, getContext(serial));
+ for(vector<ApplicationObserverPrx>::const_iterator p = _publishers.begin(); p != _publishers.end(); ++p)
+ {
+ (*p)->applicationUpdated(serial, info, getContext(serial));
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -653,14 +714,14 @@ AdapterObserverTopic::AdapterObserverTopic(const IceStorm::TopicManagerPrx& topi
ObserverTopic(topicManager, "AdapterObserver"),
_adapters(adapters)
{
- const_cast<AdapterObserverPrx&>(_publisher) = AdapterObserverPrx::uncheckedCast(_basePublisher);
+ _publishers = getPublishers<AdapterObserverPrx>();
}
int
AdapterObserverTopic::adapterInit(const AdapterInfoSeq& adpts)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return -1;
}
@@ -672,7 +733,10 @@ AdapterObserverTopic::adapterInit(const AdapterInfoSeq& adpts)
}
try
{
- _publisher->adapterInit(adpts, getContext(_serial));
+ for(vector<AdapterObserverPrx>::const_iterator p = _publishers.begin(); p != _publishers.end(); ++p)
+ {
+ (*p)->adapterInit(adpts, getContext(_serial));
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -687,7 +751,7 @@ int
AdapterObserverTopic::adapterAdded(const AdapterInfo& info)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return -1;
}
@@ -695,7 +759,10 @@ AdapterObserverTopic::adapterAdded(const AdapterInfo& info)
_adapters.insert(make_pair(info.id, info));
try
{
- _publisher->adapterAdded(info, getContext(_serial));
+ for(vector<AdapterObserverPrx>::const_iterator p = _publishers.begin(); p != _publishers.end(); ++p)
+ {
+ (*p)->adapterAdded(info, getContext(_serial));
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -710,7 +777,7 @@ int
AdapterObserverTopic::adapterUpdated(const AdapterInfo& info)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return -1;
}
@@ -718,7 +785,10 @@ AdapterObserverTopic::adapterUpdated(const AdapterInfo& info)
_adapters[info.id] = info;
try
{
- _publisher->adapterUpdated(info, getContext(_serial));
+ for(vector<AdapterObserverPrx>::const_iterator p = _publishers.begin(); p != _publishers.end(); ++p)
+ {
+ (*p)->adapterUpdated(info, getContext(_serial));
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -733,7 +803,7 @@ int
AdapterObserverTopic::adapterRemoved(const string& id)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return -1;
}
@@ -741,7 +811,10 @@ AdapterObserverTopic::adapterRemoved(const string& id)
_adapters.erase(id);
try
{
- _publisher->adapterRemoved(id, getContext(_serial));
+ for(vector<AdapterObserverPrx>::const_iterator p = _publishers.begin(); p != _publishers.end(); ++p)
+ {
+ (*p)->adapterRemoved(id, getContext(_serial));
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -769,14 +842,14 @@ ObjectObserverTopic::ObjectObserverTopic(const IceStorm::TopicManagerPrx& topicM
ObserverTopic(topicManager, "ObjectObserver"),
_objects(objects)
{
- const_cast<ObjectObserverPrx&>(_publisher) = ObjectObserverPrx::uncheckedCast(_basePublisher);
+ _publishers = getPublishers<ObjectObserverPrx>();
}
int
ObjectObserverTopic::objectInit(const ObjectInfoSeq& objects)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return -1;
}
@@ -788,7 +861,10 @@ ObjectObserverTopic::objectInit(const ObjectInfoSeq& objects)
}
try
{
- _publisher->objectInit(objects, getContext(_serial));
+ for(vector<ObjectObserverPrx>::const_iterator p = _publishers.begin(); p != _publishers.end(); ++p)
+ {
+ (*p)->objectInit(objects, getContext(_serial));
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -803,7 +879,7 @@ int
ObjectObserverTopic::objectAdded(const ObjectInfo& info)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return -1;
}
@@ -811,7 +887,10 @@ ObjectObserverTopic::objectAdded(const ObjectInfo& info)
_objects.insert(make_pair(info.proxy->ice_getIdentity(), info));
try
{
- _publisher->objectAdded(info, getContext(_serial));
+ for(vector<ObjectObserverPrx>::const_iterator p = _publishers.begin(); p != _publishers.end(); ++p)
+ {
+ (*p)->objectAdded(info, getContext(_serial));
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -826,7 +905,7 @@ int
ObjectObserverTopic::objectUpdated(const ObjectInfo& info)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return -1;
}
@@ -834,7 +913,10 @@ ObjectObserverTopic::objectUpdated(const ObjectInfo& info)
_objects[info.proxy->ice_getIdentity()] = info;
try
{
- _publisher->objectUpdated(info, getContext(_serial));
+ for(vector<ObjectObserverPrx>::const_iterator p = _publishers.begin(); p != _publishers.end(); ++p)
+ {
+ (*p)->objectUpdated(info, getContext(_serial));
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -849,7 +931,7 @@ int
ObjectObserverTopic::objectRemoved(const Ice::Identity& id)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return -1;
}
@@ -857,7 +939,10 @@ ObjectObserverTopic::objectRemoved(const Ice::Identity& id)
_objects.erase(id);
try
{
- _publisher->objectRemoved(id, getContext(_serial));
+ for(vector<ObjectObserverPrx>::const_iterator p = _publishers.begin(); p != _publishers.end(); ++p)
+ {
+ (*p)->objectRemoved(id, getContext(_serial));
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -872,7 +957,7 @@ int
ObjectObserverTopic::objectsAddedOrUpdated(const ObjectInfoSeq& infos)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return -1;
}
@@ -886,7 +971,10 @@ ObjectObserverTopic::objectsAddedOrUpdated(const ObjectInfoSeq& infos)
q->second = *p;
try
{
- _publisher->objectUpdated(*p, getContext(_serial));
+ for(vector<ObjectObserverPrx>::const_iterator q = _publishers.begin(); q != _publishers.end(); ++q)
+ {
+ (*q)->objectUpdated(*p, getContext(_serial));
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -899,7 +987,10 @@ ObjectObserverTopic::objectsAddedOrUpdated(const ObjectInfoSeq& infos)
_objects.insert(make_pair(p->proxy->ice_getIdentity(), *p));
try
{
- _publisher->objectAdded(*p, getContext(_serial));
+ for(vector<ObjectObserverPrx>::const_iterator q = _publishers.begin(); q != _publishers.end(); ++q)
+ {
+ (*q)->objectAdded(*p, getContext(_serial));
+ }
}
catch(const Ice::LocalException& ex)
{
@@ -922,7 +1013,7 @@ int
ObjectObserverTopic::objectsRemoved(const ObjectInfoSeq& infos)
{
Lock sync(*this);
- if(!_topic)
+ if(_topics.empty())
{
return -1;
}
@@ -933,7 +1024,10 @@ ObjectObserverTopic::objectsRemoved(const ObjectInfoSeq& infos)
_objects.erase(p->proxy->ice_getIdentity());
try
{
- _publisher->objectRemoved(p->proxy->ice_getIdentity(), getContext(_serial));
+ for(vector<ObjectObserverPrx>::const_iterator q = _publishers.begin(); q != _publishers.end(); ++q)
+ {
+ (*q)->objectRemoved(p->proxy->ice_getIdentity(), getContext(_serial));
+ }
}
catch(const Ice::LocalException& ex)
{
diff --git a/cpp/src/IceGrid/Topics.h b/cpp/src/IceGrid/Topics.h
index 7399457d35e..1a27af89ae6 100644
--- a/cpp/src/IceGrid/Topics.h
+++ b/cpp/src/IceGrid/Topics.h
@@ -43,9 +43,19 @@ protected:
void updateSerial(int);
Ice::Context getContext(int) const;
+ template<typename T> std::vector<T> getPublishers() const
+ {
+ std::vector<T> publishers;
+ for(std::vector<Ice::ObjectPrx>::const_iterator p = _basePublishers.begin(); p != _basePublishers.end(); ++p)
+ {
+ publishers.push_back(T::uncheckedCast(*p));
+ }
+ return publishers;
+ }
+
Ice::LoggerPtr _logger;
- IceStorm::TopicPrx _topic;
- Ice::ObjectPrx _basePublisher;
+ std::map<Ice::EncodingVersion, IceStorm::TopicPrx> _topics;
+ std::vector<Ice::ObjectPrx> _basePublishers;
int _serial;
std::set<std::string> _syncSubscribers;
@@ -67,7 +77,7 @@ public:
private:
- const RegistryObserverPrx _publisher;
+ std::vector<RegistryObserverPrx> _publishers;
std::map<std::string, RegistryInfo> _registries;
};
typedef IceUtil::Handle<RegistryObserverTopic> RegistryObserverTopicPtr;
@@ -92,7 +102,7 @@ public:
private:
const NodeObserverPrx _externalPublisher;
- const NodeObserverPrx _publisher;
+ std::vector<NodeObserverPrx> _publishers;
std::map<std::string, NodeDynamicInfo> _nodes;
};
typedef IceUtil::Handle<NodeObserverTopic> NodeObserverTopicPtr;
@@ -112,7 +122,7 @@ public:
private:
- const ApplicationObserverPrx _publisher;
+ std::vector<ApplicationObserverPrx> _publishers;
std::map<std::string, ApplicationInfo> _applications;
};
typedef IceUtil::Handle<ApplicationObserverTopic> ApplicationObserverTopicPtr;
@@ -132,7 +142,7 @@ public:
private:
- const AdapterObserverPrx _publisher;
+ std::vector<AdapterObserverPrx> _publishers;
std::map<std::string, AdapterInfo> _adapters;
};
typedef IceUtil::Handle<AdapterObserverTopic> AdapterObserverTopicPtr;
@@ -155,7 +165,7 @@ public:
private:
- const ObjectObserverPrx _publisher;
+ std::vector<ObjectObserverPrx> _publishers;
std::map<Ice::Identity, ObjectInfo> _objects;
};
typedef IceUtil::Handle<ObjectObserverTopic> ObjectObserverTopicPtr;
generated by cgit v1.2.3 (git 2.46.0) at 2025-08-04 16:24:12 +0000