Removed ice_flush() from proxy base class and implemented

    Communicator::flushBatchRequests().

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