summaryrefslogtreecommitdiff
path: root/cpp/src
diff options
context:
space:
mode:
Diffstat (limited to 'cpp/src')
-rw-r--r--cpp/src/Glacier/Request.cpp7
-rw-r--r--cpp/src/Ice/CommunicatorI.cpp2
-rw-r--r--cpp/src/Ice/ConnectionFactory.cpp24
-rw-r--r--cpp/src/Ice/ConnectionFactory.h2
-rw-r--r--cpp/src/Ice/Instance.cpp18
-rw-r--r--cpp/src/Ice/Instance.h1
-rw-r--r--cpp/src/Ice/ObjectAdapterI.cpp11
-rw-r--r--cpp/src/Ice/ObjectAdapterI.h2
-rw-r--r--cpp/src/Ice/Proxy.cpp39
-rw-r--r--cpp/src/IceStorm/Flusher.cpp6
-rw-r--r--cpp/src/IceStorm/LinkSubscriber.cpp33
-rw-r--r--cpp/src/IceStorm/LinkSubscriber.h4
-rw-r--r--cpp/src/IceStorm/OnewayBatchSubscriber.cpp31
-rw-r--r--cpp/src/IceStorm/OnewayBatchSubscriber.h4
-rw-r--r--cpp/src/IceStorm/SubscriberFactory.cpp9
-rw-r--r--cpp/src/IceStorm/SubscriberFactory.h3
-rw-r--r--cpp/src/IceStorm/TopicManagerI.cpp2
17 files changed, 90 insertions, 108 deletions
diff --git a/cpp/src/Glacier/Request.cpp b/cpp/src/Glacier/Request.cpp
index ea8867d9752..f3186799d91 100644
--- a/cpp/src/Glacier/Request.cpp
+++ b/cpp/src/Glacier/Request.cpp
@@ -289,12 +289,7 @@ Glacier::RequestQueue::run()
//
// This sends all batched missives.
//
-// sort(proxies.begin(), proxies.end());
-// proxies.erase(unique(proxies.begin(), proxies.end()), proxies.end());
- for(vector<ObjectPrx>::const_iterator q = proxies.begin(); q != proxies.end(); ++q)
- {
- (*q)->ice_flush();
- }
+ _communicator->flushBatchRequests();
}
catch(const Ice::Exception& ex)
{
diff --git a/cpp/src/Ice/CommunicatorI.cpp b/cpp/src/Ice/CommunicatorI.cpp
index 4a156a0fa3f..731576cb23c 100644
--- a/cpp/src/Ice/CommunicatorI.cpp
+++ b/cpp/src/Ice/CommunicatorI.cpp
@@ -260,7 +260,7 @@ Ice::CommunicatorI::getPluginManager()
void
Ice::CommunicatorI::flushBatchRequests()
{
- // TODO: implement this.
+ _instance->flushBatchRequests();
}
Ice::CommunicatorI::CommunicatorI(int& argc, char* argv[], const PropertiesPtr& properties) :
diff --git a/cpp/src/Ice/ConnectionFactory.cpp b/cpp/src/Ice/ConnectionFactory.cpp
index 6661b7322b2..ae99f562611 100644
--- a/cpp/src/Ice/ConnectionFactory.cpp
+++ b/cpp/src/Ice/ConnectionFactory.cpp
@@ -380,6 +380,23 @@ IceInternal::OutgoingConnectionFactory::removeAdapter(const ObjectAdapterPtr& ad
}
}
+void
+IceInternal::OutgoingConnectionFactory::flushBatchRequests()
+{
+ list<ConnectionPtr> c;
+ {
+ IceUtil::Monitor<IceUtil::Mutex>::Lock sync(*this);
+
+ for(std::multimap<EndpointPtr, ConnectionPtr>::const_iterator p = _connections.begin();
+ p != _connections.end();
+ ++p)
+ {
+ c.push_back(p->second);
+ }
+ }
+ for_each(c.begin(), c.end(), Ice::voidMemFun(&Connection::flushBatchRequest));
+}
+
IceInternal::OutgoingConnectionFactory::OutgoingConnectionFactory(const InstancePtr& instance) :
_instance(instance),
_destroyed(false)
@@ -493,6 +510,13 @@ IceInternal::IncomingConnectionFactory::connections() const
return result;
}
+void
+IceInternal::IncomingConnectionFactory::flushBatchRequests()
+{
+ list<ConnectionPtr> c = connections(); // connections() is synchronized, so need to synchronize here.
+ for_each(c.begin(), c.end(), Ice::voidMemFun(&Connection::flushBatchRequest));
+}
+
bool
IceInternal::IncomingConnectionFactory::datagram() const
{
diff --git a/cpp/src/Ice/ConnectionFactory.h b/cpp/src/Ice/ConnectionFactory.h
index 1f0cfb9ea2f..d79ad31cd74 100644
--- a/cpp/src/Ice/ConnectionFactory.h
+++ b/cpp/src/Ice/ConnectionFactory.h
@@ -51,6 +51,7 @@ public:
ConnectionPtr create(const std::vector<EndpointPtr>&);
void setRouter(const ::Ice::RouterPrx&);
void removeAdapter(const ::Ice::ObjectAdapterPtr&);
+ void flushBatchRequests();
private:
@@ -78,6 +79,7 @@ public:
EndpointPtr endpoint() const;
bool equivalent(const EndpointPtr&) const;
std::list<ConnectionPtr> connections() const;
+ void flushBatchRequests();
//
// Operations from EventHandler
diff --git a/cpp/src/Ice/Instance.cpp b/cpp/src/Ice/Instance.cpp
index e2ad598d146..52ba8ea430d 100644
--- a/cpp/src/Ice/Instance.cpp
+++ b/cpp/src/Ice/Instance.cpp
@@ -360,6 +360,24 @@ IceInternal::Instance::messageSizeMax() const
return _messageSizeMax;
}
+void
+IceInternal::Instance::flushBatchRequests()
+{
+ OutgoingConnectionFactoryPtr factory;
+ std::map<std::string, ::Ice::ObjectAdapterIPtr> adapters;
+ {
+ IceUtil::RecMutex::Lock sync(*this);
+
+ factory = _outgoingConnectionFactory;
+ adapters = _objectAdapterFactory->_adapters;
+ }
+ factory->flushBatchRequests();
+ for(std::map<std::string, ::Ice::ObjectAdapterIPtr>::const_iterator p = adapters.begin(); p != adapters.end(); ++p)
+ {
+ p->second->flushBatchRequests();
+ }
+}
+
IceInternal::Instance::Instance(const CommunicatorPtr& communicator, int& argc, char* argv[],
const PropertiesPtr& properties) :
_destroyed(false),
diff --git a/cpp/src/Ice/Instance.h b/cpp/src/Ice/Instance.h
index 1f919b35c68..4f31513345f 100644
--- a/cpp/src/Ice/Instance.h
+++ b/cpp/src/Ice/Instance.h
@@ -74,6 +74,7 @@ public:
DynamicLibraryListPtr dynamicLibraryList();
Ice::PluginManagerPtr pluginManager();
size_t messageSizeMax() const;
+ void flushBatchRequests();
private:
diff --git a/cpp/src/Ice/ObjectAdapterI.cpp b/cpp/src/Ice/ObjectAdapterI.cpp
index 72d58545ddf..afa026ff9aa 100644
--- a/cpp/src/Ice/ObjectAdapterI.cpp
+++ b/cpp/src/Ice/ObjectAdapterI.cpp
@@ -466,6 +466,17 @@ Ice::ObjectAdapterI::getIncomingConnections() const
}
void
+Ice::ObjectAdapterI::flushBatchRequests()
+{
+ std::vector<IceInternal::IncomingConnectionFactoryPtr> f;
+ {
+ IceUtil::Monitor<IceUtil::RecMutex>::Lock sync(*this);
+ f = _incomingConnectionFactories;
+ }
+ for_each(f.begin(), f.end(), Ice::voidMemFun(&IncomingConnectionFactory::flushBatchRequests));
+}
+
+void
Ice::ObjectAdapterI::incDirectCount()
{
IceUtil::Monitor<IceUtil::RecMutex>::Lock sync(*this);
diff --git a/cpp/src/Ice/ObjectAdapterI.h b/cpp/src/Ice/ObjectAdapterI.h
index 81564d486aa..be3c87c4692 100644
--- a/cpp/src/Ice/ObjectAdapterI.h
+++ b/cpp/src/Ice/ObjectAdapterI.h
@@ -74,6 +74,8 @@ public:
std::list<IceInternal::ConnectionPtr> getIncomingConnections() const;
+ void flushBatchRequests();
+
void incDirectCount();
void decDirectCount();
diff --git a/cpp/src/Ice/Proxy.cpp b/cpp/src/Ice/Proxy.cpp
index f43c82b2615..780cff5e38c 100644
--- a/cpp/src/Ice/Proxy.cpp
+++ b/cpp/src/Ice/Proxy.cpp
@@ -631,33 +631,6 @@ IceProxy::Ice::Object::ice_default() const
}
}
-void
-IceProxy::Ice::Object::ice_flush()
-{
- //
- // Retry is necessary for ice_flush in case the current connection
- // is closed. If that's the case we need to get a new connection.
- //
- int __cnt = 0;
- while(true)
- {
- try
- {
- Handle< ::IceDelegate::Ice::Object> __del = __getDelegate();
- __del->ice_flush();
- return;
- }
- catch(const DatagramLimitException&)
- {
- throw; // DatagramLimitExcpetion is not repeatable.
- }
- catch(const LocalException& __ex)
- {
- __handleException(__ex, __cnt);
- }
- }
-}
-
ReferencePtr
IceProxy::Ice::Object::__reference() const
{
@@ -1006,12 +979,6 @@ IceDelegateM::Ice::Object::ice_invoke_async(const AMI_Object_ice_invokePtr& cb,
}
void
-IceDelegateM::Ice::Object::ice_flush()
-{
- __connection->flushBatchRequest();
-}
-
-void
IceDelegateM::Ice::Object::__copyFrom(const ::IceInternal::Handle< ::IceDelegateM::Ice::Object>& from)
{
//
@@ -1319,12 +1286,6 @@ IceDelegateD::Ice::Object::ice_invoke_async(const AMI_Object_ice_invokePtr&,
}
void
-IceDelegateD::Ice::Object::ice_flush()
-{
- // Nothing to do for direct delegates.
-}
-
-void
IceDelegateD::Ice::Object::__copyFrom(const ::IceInternal::Handle< ::IceDelegateD::Ice::Object>& from)
{
//
diff --git a/cpp/src/IceStorm/Flusher.cpp b/cpp/src/IceStorm/Flusher.cpp
index 4ef6b3374d1..302d9160b2f 100644
--- a/cpp/src/IceStorm/Flusher.cpp
+++ b/cpp/src/IceStorm/Flusher.cpp
@@ -35,6 +35,7 @@ class FlusherThread : public IceUtil::Thread, public IceUtil::Monitor<IceUtil::M
public:
FlusherThread(const Ice::CommunicatorPtr& communicator, const TraceLevelsPtr& traceLevels) :
+ _communicator(communicator),
_traceLevels(traceLevels),
_destroy(false)
{
@@ -131,7 +132,8 @@ private:
//
_subscribers.erase(remove_if(_subscribers.begin(), _subscribers.end(),
IceUtil::constMemFun(&Flushable::inactive)), _subscribers.end());
- for_each(_subscribers.begin(), _subscribers.end(), IceUtil::voidMemFun(&Flushable::flush));
+
+ _communicator->flushBatchRequests();
//
// Trace after the flush so that the correct number of objects
@@ -150,8 +152,8 @@ private:
return (_subscribers.empty()) ? 0 : _flushTime;
}
+ Ice::CommunicatorPtr _communicator;
TraceLevelsPtr _traceLevels;
-
FlushableList _subscribers;
bool _destroy;
long _flushTime;
diff --git a/cpp/src/IceStorm/LinkSubscriber.cpp b/cpp/src/IceStorm/LinkSubscriber.cpp
index e753ad38e2b..8eb19e3476f 100644
--- a/cpp/src/IceStorm/LinkSubscriber.cpp
+++ b/cpp/src/IceStorm/LinkSubscriber.cpp
@@ -20,10 +20,10 @@
using namespace IceStorm;
using namespace std;
-LinkSubscriber::LinkSubscriber(const SubscriberFactoryPtr& factory, const TraceLevelsPtr& traceLevels,
- const QueuedProxyPtr& obj, Ice::Int cost) :
+LinkSubscriber::LinkSubscriber(const SubscriberFactoryPtr& factory, const Ice::CommunicatorPtr& communicator,
+ const TraceLevelsPtr& traceLevels, const QueuedProxyPtr& obj, Ice::Int cost) :
Subscriber(traceLevels, obj->proxy()->ice_getIdentity()),
- _factory(factory), _obj(obj), _cost(cost)
+ _factory(factory), _communicator(communicator), _obj(obj), _cost(cost)
{
_factory->incProxyUsageCount(_obj);
}
@@ -115,32 +115,7 @@ LinkSubscriber::publish(const EventPtr& event)
void
LinkSubscriber::flush()
{
- try
- {
- _obj->proxy()->ice_flush();
- }
- catch(const Ice::ObjectNotExistException& e)
- {
- //
- // ObjectNotExist causes the link to be removed.
- //
- IceUtil::Mutex::Lock sync(_stateMutex);
- _state = StateError;
-
- if(_traceLevels->subscriber > 0)
- {
- Ice::Trace out(_traceLevels->logger, _traceLevels->subscriberCat);
- out << id() << ": link topic flush failed: " << e;
- }
- }
- catch(const Ice::LocalException& e)
- {
- if(_traceLevels->subscriber > 0)
- {
- Ice::Trace out(_traceLevels->logger, _traceLevels->subscriberCat);
- out << id() << ": link topic flush failed: " << e;
- }
- }
+ _communicator->flushBatchRequests();
}
bool
diff --git a/cpp/src/IceStorm/LinkSubscriber.h b/cpp/src/IceStorm/LinkSubscriber.h
index 25dc7b6ae96..f5eacff632e 100644
--- a/cpp/src/IceStorm/LinkSubscriber.h
+++ b/cpp/src/IceStorm/LinkSubscriber.h
@@ -30,7 +30,8 @@ class LinkSubscriber : public Subscriber, public Flushable
{
public:
- LinkSubscriber(const SubscriberFactoryPtr&, const TraceLevelsPtr&, const QueuedProxyPtr&, Ice::Int);
+ LinkSubscriber(const SubscriberFactoryPtr&, const Ice::CommunicatorPtr&,
+ const TraceLevelsPtr&, const QueuedProxyPtr&, Ice::Int);
~LinkSubscriber();
virtual bool persistent() const;
@@ -46,6 +47,7 @@ private:
// Immutable
SubscriberFactoryPtr _factory;
+ Ice::CommunicatorPtr _communicator;
QueuedProxyPtr _obj;
Ice::Int _cost;
};
diff --git a/cpp/src/IceStorm/OnewayBatchSubscriber.cpp b/cpp/src/IceStorm/OnewayBatchSubscriber.cpp
index d54d223f1f9..e8869cd8e13 100644
--- a/cpp/src/IceStorm/OnewayBatchSubscriber.cpp
+++ b/cpp/src/IceStorm/OnewayBatchSubscriber.cpp
@@ -20,9 +20,13 @@
using namespace IceStorm;
using namespace std;
-OnewayBatchSubscriber::OnewayBatchSubscriber(const SubscriberFactoryPtr& factory, const TraceLevelsPtr& traceLevels,
- const FlusherPtr& flusher, const QueuedProxyPtr& obj) :
+OnewayBatchSubscriber::OnewayBatchSubscriber(const SubscriberFactoryPtr& factory,
+ const Ice::CommunicatorPtr& communicator,
+ const TraceLevelsPtr& traceLevels,
+ const FlusherPtr& flusher,
+ const QueuedProxyPtr& obj) :
OnewaySubscriber(factory, traceLevels, obj),
+ _communicator(communicator),
_flusher(flusher)
{
_flusher->add(this);
@@ -79,28 +83,7 @@ OnewayBatchSubscriber::inactive() const
void
OnewayBatchSubscriber::flush()
{
- try
- {
- _obj->proxy()->ice_flush();
- }
- catch(const Ice::LocalException& e)
- {
- IceUtil::Mutex::Lock sync(_stateMutex);
- //
- // It's possible that the subscriber was unsubscribed, or
- // marked invalid by another thread. Don't display a
- // diagnostic in this case.
- //
- if(_state == StateActive)
- {
- if(_traceLevels->subscriber > 0)
- {
- Ice::Trace out(_traceLevels->logger, _traceLevels->subscriberCat);
- out << id() << ": flush failed: " << e;
- }
- _state = StateError;
- }
- }
+ _communicator->flushBatchRequests();
}
bool
diff --git a/cpp/src/IceStorm/OnewayBatchSubscriber.h b/cpp/src/IceStorm/OnewayBatchSubscriber.h
index 2c209551f80..4aef9f36a35 100644
--- a/cpp/src/IceStorm/OnewayBatchSubscriber.h
+++ b/cpp/src/IceStorm/OnewayBatchSubscriber.h
@@ -32,7 +32,8 @@ class OnewayBatchSubscriber : public OnewaySubscriber, public Flushable
{
public:
- OnewayBatchSubscriber(const SubscriberFactoryPtr&, const TraceLevelsPtr&, const FlusherPtr&, const QueuedProxyPtr&);
+ OnewayBatchSubscriber(const SubscriberFactoryPtr&, const Ice::CommunicatorPtr&,
+ const TraceLevelsPtr&, const FlusherPtr&, const QueuedProxyPtr&);
~OnewayBatchSubscriber();
virtual void unsubscribe();
@@ -45,6 +46,7 @@ public:
private:
+ Ice::CommunicatorPtr _communicator;
FlusherPtr _flusher;
};
diff --git a/cpp/src/IceStorm/SubscriberFactory.cpp b/cpp/src/IceStorm/SubscriberFactory.cpp
index 8ce91506925..4aabd860592 100644
--- a/cpp/src/IceStorm/SubscriberFactory.cpp
+++ b/cpp/src/IceStorm/SubscriberFactory.cpp
@@ -25,7 +25,10 @@
using namespace std;
using namespace IceStorm;
-SubscriberFactory::SubscriberFactory(const TraceLevelsPtr& traceLevels, const FlusherPtr& flusher) :
+SubscriberFactory::SubscriberFactory(const Ice::CommunicatorPtr& communicator,
+ const TraceLevelsPtr& traceLevels,
+ const FlusherPtr& flusher) :
+ _communicator(communicator),
_traceLevels(traceLevels),
_flusher(flusher)
{
@@ -59,7 +62,7 @@ SubscriberFactory::createLinkSubscriber(const TopicLinkPrx& obj, Ice::Int cost)
_proxies.insert(pair<Ice::ObjectPrx, ProxyInfo>(newObj, info));
}
- return new LinkSubscriber(this, _traceLevels, proxy, cost);
+ return new LinkSubscriber(this, _communicator, _traceLevels, proxy, cost);
}
SubscriberPtr
@@ -134,7 +137,7 @@ SubscriberFactory::createSubscriber(const QoS& qos, const Ice::ObjectPrx& obj)
if(reliability == "batch")
{
- return new OnewayBatchSubscriber(this, _traceLevels, _flusher, proxy);
+ return new OnewayBatchSubscriber(this, _communicator, _traceLevels, _flusher, proxy);
}
else
{
diff --git a/cpp/src/IceStorm/SubscriberFactory.h b/cpp/src/IceStorm/SubscriberFactory.h
index 6c653d31b7d..eeabc8d29a2 100644
--- a/cpp/src/IceStorm/SubscriberFactory.h
+++ b/cpp/src/IceStorm/SubscriberFactory.h
@@ -43,7 +43,7 @@ class SubscriberFactory : public IceUtil::Shared
{
public:
- SubscriberFactory(const TraceLevelsPtr&, const FlusherPtr&);
+ SubscriberFactory(const Ice::CommunicatorPtr&, const TraceLevelsPtr&, const FlusherPtr&);
//
// Create a link subscriber (that is a subscriber that points to
@@ -79,6 +79,7 @@ private:
Ice::Int count;
};
+ Ice::CommunicatorPtr _communicator;
TraceLevelsPtr _traceLevels;
FlusherPtr _flusher;
IceUtil::RecMutex _proxiesMutex;
diff --git a/cpp/src/IceStorm/TopicManagerI.cpp b/cpp/src/IceStorm/TopicManagerI.cpp
index 2d17084902b..009574b7706 100644
--- a/cpp/src/IceStorm/TopicManagerI.cpp
+++ b/cpp/src/IceStorm/TopicManagerI.cpp
@@ -35,7 +35,7 @@ TopicManagerI::TopicManagerI(const Ice::CommunicatorPtr& communicator, const Ice
_topics(_communicator, envName, dbName)
{
_flusher = new Flusher(_communicator, _traceLevels);
- _factory = new SubscriberFactory(_traceLevels, _flusher);
+ _factory = new SubscriberFactory(_communicator, _traceLevels, _flusher);
//
// Recreate each of the topics in the dictionary. If the topic
generated by cgit v1.2.3 (git 2.46.0) at 2025-08-04 15:34:46 +0000