Fixed ICE-5486: added support for MessageSizeMax OA property

12 files changed, 90 insertions, 25 deletions

diff --git a/cpp/src/Ice/ConnectionFactory.cpp b/cpp/src/Ice/ConnectionFactory.cpp
index 154e1ef193b..db378242b6a 100644
--- a/cpp/src/Ice/ConnectionFactory.cpp
+++ b/cpp/src/Ice/ConnectionFactory.cpp
@@ -1603,7 +1603,7 @@ IceInternal::IncomingConnectionFactory::connectionStartFailed(const Ice::Connect
 //
 IceInternal::IncomingConnectionFactory::IncomingConnectionFactory(const InstancePtr& instance,
                                                                   const EndpointIPtr& endpoint,
-                                                                  const ObjectAdapterPtr& adapter) :
+                                                                  const ObjectAdapterIPtr& adapter) :
     _instance(instance),
     _monitor(new FactoryACMMonitor(instance, dynamic_cast<ObjectAdapterI*>(adapter.get())->getACM())),
     _endpoint(endpoint),
@@ -1664,7 +1664,7 @@ IceInternal::IncomingConnectionFactory::initialize(const string& oaName)
                 out << "listening for " << _endpoint->protocol() << " connections\n" << _acceptor->toDetailedString();
             }
 
-            dynamic_cast<ObjectAdapterI*>(_adapter.get())->getThreadPool()->initialize(this);
+            _adapter->getThreadPool()->initialize(this);
         }
     }
     catch(const Ice::Exception&)
@@ -1730,7 +1730,7 @@ IceInternal::IncomingConnectionFactory::setState(State state)
                     Trace out(_instance->initializationData().logger, _instance->traceLevels()->networkCat);
                     out << "accepting " << _endpoint->protocol() << " connections at " << _acceptor->toString();
                 }
-                dynamic_cast<ObjectAdapterI*>(_adapter.get())->getThreadPool()->_register(this, SocketOperationRead);
+                _adapter->getThreadPool()->_register(this, SocketOperationRead);
             }
             for_each(_connections.begin(), _connections.end(), Ice::voidMemFun(&ConnectionI::activate));
             break;
@@ -1749,7 +1749,7 @@ IceInternal::IncomingConnectionFactory::setState(State state)
                     Trace out(_instance->initializationData().logger, _instance->traceLevels()->networkCat);
                     out << "holding " << _endpoint->protocol() << " connections at " << _acceptor->toString();
                 }
-                dynamic_cast<ObjectAdapterI*>(_adapter.get())->getThreadPool()->unregister(this, SocketOperationRead);
+                _adapter->getThreadPool()->unregister(this, SocketOperationRead);
             }
             for_each(_connections.begin(), _connections.end(), Ice::voidMemFun(&ConnectionI::hold));
             break;
@@ -1766,7 +1766,7 @@ IceInternal::IncomingConnectionFactory::setState(State state)
                 // the finish() call. Not all selector implementations do support this 
                 // however.
                 //
-                if(dynamic_cast<ObjectAdapterI*>(_adapter.get())->getThreadPool()->finish(this, true))
+                if(_adapter->getThreadPool()->finish(this, true))
                 {
                     closeAcceptor(true);
                 }
diff --git a/cpp/src/Ice/ConnectionFactory.h b/cpp/src/Ice/ConnectionFactory.h
index c6b688225b5..1026027d87e 100644
--- a/cpp/src/Ice/ConnectionFactory.h
+++ b/cpp/src/Ice/ConnectionFactory.h
@@ -36,6 +36,7 @@ namespace Ice
 
 class LocalException;
 class ObjectAdapterI;
+typedef IceUtil::Handle<ObjectAdapterI> ObjectAdapterIPtr;
 
 }
 
@@ -199,7 +200,7 @@ public:
 
 private:
 
-    IncomingConnectionFactory(const InstancePtr&, const EndpointIPtr&, const Ice::ObjectAdapterPtr&);
+    IncomingConnectionFactory(const InstancePtr&, const EndpointIPtr&, const Ice::ObjectAdapterIPtr&);
     void initialize(const std::string&);
     virtual ~IncomingConnectionFactory();
     friend class Ice::ObjectAdapterI;
@@ -222,7 +223,7 @@ private:
     const TransceiverPtr _transceiver;
     const EndpointIPtr _endpoint;
 
-    Ice::ObjectAdapterPtr _adapter;
+    Ice::ObjectAdapterIPtr _adapter;
 
     const bool _warn;
 
diff --git a/cpp/src/Ice/ConnectionI.cpp b/cpp/src/Ice/ConnectionI.cpp
index 36bb80fc952..2a3077aa7b8 100644
--- a/cpp/src/Ice/ConnectionI.cpp
+++ b/cpp/src/Ice/ConnectionI.cpp
@@ -1845,9 +1845,9 @@ Ice::ConnectionI::message(ThreadPoolCurrent& current)
                     {
                         throw IllegalMessageSizeException(__FILE__, __LINE__);
                     }
-                    if(size > static_cast<Int>(_instance->messageSizeMax()))
+                    if(size > static_cast<Int>(_messageSizeMax))
                     {
-                        Ex::throwMemoryLimitException(__FILE__, __LINE__, size, _instance->messageSizeMax());
+                        Ex::throwMemoryLimitException(__FILE__, __LINE__, size, _messageSizeMax);
                     }
                     if(size > static_cast<Int>(_readStream.b.size()))
                     {
@@ -2380,7 +2380,7 @@ Ice::ConnectionI::ConnectionI(const CommunicatorPtr& communicator,
                               const TransceiverPtr& transceiver,
                               const ConnectorPtr& connector,
                               const EndpointIPtr& endpoint,
-                              const ObjectAdapterPtr& adapter) :
+                              const ObjectAdapterIPtr& adapter) :
     _communicator(communicator),
     _instance(instance),
     _monitor(monitor),
@@ -2404,6 +2404,7 @@ Ice::ConnectionI::ConnectionI(const CommunicatorPtr& communicator,
     _nextRequestId(1),
     _requestsHint(_requests.end()),
     _asyncRequestsHint(_asyncRequests.end()),
+    _messageSizeMax(adapter ? adapter->messageSizeMax() : _instance->messageSizeMax()),
     _batchAutoFlushSize(_instance->batchAutoFlushSize()),
     _batchStream(_instance.get(), Ice::currentProtocolEncoding),
     _batchStreamInUse(false),
@@ -2432,10 +2433,9 @@ Ice::ConnectionI::ConnectionI(const CommunicatorPtr& communicator,
         compressionLevel = 9;
     }
 
-    ObjectAdapterI* adapterImpl = _adapter ? dynamic_cast<ObjectAdapterI*>(_adapter.get()) : 0;
-    if(adapterImpl)
+    if(adapter)
     {
-        _servantManager = adapterImpl->getServantManager();
+        _servantManager = adapter->getServantManager();
     }
 
     if(_monitor && _monitor->getACM().timeout > 0)
@@ -2446,9 +2446,9 @@ Ice::ConnectionI::ConnectionI(const CommunicatorPtr& communicator,
     __setNoDelete(true);
     try
     {
-        if(adapterImpl)
+        if(adapter)
         {
-            const_cast<ThreadPoolPtr&>(_threadPool) = adapterImpl->getThreadPool();
+            const_cast<ThreadPoolPtr&>(_threadPool) = adapter->getThreadPool();
         }
         else
         {
@@ -3380,9 +3380,9 @@ Ice::ConnectionI::doUncompress(BasicStream& compressed, BasicStream& uncompresse
         throw IllegalMessageSizeException(__FILE__, __LINE__);
     }
 
-    if(uncompressedSize > static_cast<Int>(_instance->messageSizeMax()))
+    if(uncompressedSize > static_cast<Int>(_messageSizeMax))
     {
-        Ex::throwMemoryLimitException(__FILE__, __LINE__, uncompressedSize, _instance->messageSizeMax());
+        Ex::throwMemoryLimitException(__FILE__, __LINE__, uncompressedSize, _messageSizeMax);
     }
     uncompressed.resize(uncompressedSize);
 
diff --git a/cpp/src/Ice/ConnectionI.h b/cpp/src/Ice/ConnectionI.h
index 6432863185c..003ef41c519 100644
--- a/cpp/src/Ice/ConnectionI.h
+++ b/cpp/src/Ice/ConnectionI.h
@@ -52,6 +52,8 @@ namespace Ice
 {
 
 class LocalException;
+class ObjectAdapterI;
+typedef IceUtil::Handle<ObjectAdapterI> ObjectAdapterIPtr;
 
 class ConnectionI : public Connection,
                     public IceInternal::EventHandler,
@@ -247,7 +249,7 @@ private:
 
     ConnectionI(const Ice::CommunicatorPtr&, const IceInternal::InstancePtr&, const IceInternal::ACMMonitorPtr&,
                 const IceInternal::TransceiverPtr&, const IceInternal::ConnectorPtr&,
-                const IceInternal::EndpointIPtr&, const ObjectAdapterPtr&);
+                const IceInternal::EndpointIPtr&, const ObjectAdapterIPtr&);
     virtual ~ConnectionI();
 
     friend class IceInternal::IncomingConnectionFactory;
@@ -331,6 +333,7 @@ private:
 
     IceUtil::UniquePtr<LocalException> _exception;
 
+    const size_t _messageSizeMax;
     const size_t _batchAutoFlushSize;
     IceInternal::BasicStream _batchStream;
     bool _batchStreamInUse;
diff --git a/cpp/src/Ice/Instance.cpp b/cpp/src/Ice/Instance.cpp
index 11d27fc1111..5f9034be027 100644
--- a/cpp/src/Ice/Instance.cpp
+++ b/cpp/src/Ice/Instance.cpp
@@ -1260,11 +1260,7 @@ IceInternal::Instance::Instance(const CommunicatorPtr& communicator, const Initi
         {
             static const int defaultMessageSizeMax = 1024;
             Int num = _initData.properties->getPropertyAsIntWithDefault("Ice.MessageSizeMax", defaultMessageSizeMax);
-            if(num < 1)
-            {
-                const_cast<size_t&>(_messageSizeMax) = defaultMessageSizeMax * 1024; // Ignore non-sensical values.
-            }
-            else if(static_cast<size_t>(num) > static_cast<size_t>(0x7fffffff / 1024))
+            if(num < 1 || static_cast<size_t>(num) > static_cast<size_t>(0x7fffffff / 1024))
             {
                 const_cast<size_t&>(_messageSizeMax) = static_cast<size_t>(0x7fffffff);
             }
diff --git a/cpp/src/Ice/ObjectAdapterI.cpp b/cpp/src/Ice/ObjectAdapterI.cpp
index 537aab5f79d..2fd1f92d10f 100644
--- a/cpp/src/Ice/ObjectAdapterI.cpp
+++ b/cpp/src/Ice/ObjectAdapterI.cpp
@@ -858,7 +858,8 @@ Ice::ObjectAdapterI::ObjectAdapterI(const InstancePtr& instance, const Communica
     _servantManager(new ServantManager(instance, name)),
     _name(name),
     _directCount(0),
-    _noConfig(noConfig)
+    _noConfig(noConfig),
+    _messageSizeMax(0)
 {
 }
 
@@ -923,6 +924,19 @@ Ice::ObjectAdapterI::initialize(const RouterPrx& router)
         const_cast<ACMConfig&>(_acm) = 
             ACMConfig(properties, _communicator->getLogger(), _name + ".ACM", _instance->serverACM());
 
+        {
+            const int defaultMessageSizeMax = _instance->messageSizeMax() / 1024;
+            Int num = properties->getPropertyAsIntWithDefault(_name + ".MessageSizeMax", defaultMessageSizeMax);
+            if(num < 1 || static_cast<size_t>(num) > static_cast<size_t>(0x7fffffff / 1024))
+            {
+                const_cast<size_t&>(_messageSizeMax) = static_cast<size_t>(0x7fffffff);
+            }
+            else
+            {
+                const_cast<size_t&>(_messageSizeMax) = static_cast<size_t>(num) * 1024;
+            }
+        }
+
         int threadPoolSize = properties->getPropertyAsInt(_name + ".ThreadPool.Size");
         int threadPoolSizeMax = properties->getPropertyAsInt(_name + ".ThreadPool.SizeMax");
         bool hasPriority = properties->getProperty(_name + ".ThreadPool.ThreadPriority") != "";
@@ -1415,6 +1429,7 @@ Ice::ObjectAdapterI::filterProperties(StringSeq& unknownProps)
         "Locator.PreferSecure",
         "Locator.CollocationOptimized",
         "Locator.Router",
+        "MessageSizeMax",
         "PublishedEndpoints",
         "RegisterProcess",
         "ReplicaGroupId",
diff --git a/cpp/src/Ice/ObjectAdapterI.h b/cpp/src/Ice/ObjectAdapterI.h
index a1f8711c068..25bf9830864 100644
--- a/cpp/src/Ice/ObjectAdapterI.h
+++ b/cpp/src/Ice/ObjectAdapterI.h
@@ -98,6 +98,7 @@ public:
     IceInternal::ThreadPoolPtr getThreadPool() const;
     IceInternal::ServantManagerPtr getServantManager() const;
     IceInternal::ACMConfig getACM() const;
+    size_t messageSizeMax() const { return _messageSizeMax; }
 
 private:
 
@@ -146,6 +147,7 @@ private:
     int _directCount; // The number of direct proxies dispatching on this object adapter.
     bool _noConfig;
     Identity _processId;
+    size_t _messageSizeMax;
 };
 
 }
diff --git a/cpp/src/IceGrid/RegistryI.cpp b/cpp/src/IceGrid/RegistryI.cpp
index e49f0cf49bf..5d3a73e24a7 100644
--- a/cpp/src/IceGrid/RegistryI.cpp
+++ b/cpp/src/IceGrid/RegistryI.cpp
@@ -242,6 +242,10 @@ RegistryI::startImpl()
     properties->setProperty("IceGrid.Registry.SessionManager.AdapterId", "");
     properties->setProperty("IceGrid.Registry.AdminSessionManager.AdapterId", "");
     properties->setProperty("IceGrid.Registry.Internal.AdapterId", "");
+    if(properties->getProperty("IceGrid.Registry.Internal.MessageSizeMax").empty())
+    {
+        properties->setProperty("IceGrid.Registry.Internal.MessageSizeMax", "0"); // No limit on internal data exchanged
+    }
 
     setupThreadPool(properties, "IceGrid.Registry.Client.ThreadPool", 1, 10);
     setupThreadPool(properties, "IceGrid.Registry.Server.ThreadPool", 1, 10, true); // Serialize for admin callbacks
diff --git a/cpp/src/IceStorm/Service.cpp b/cpp/src/IceStorm/Service.cpp
index 376277e9f93..0be7c03f06c 100644
--- a/cpp/src/IceStorm/Service.cpp
+++ b/cpp/src/IceStorm/Service.cpp
@@ -331,6 +331,10 @@ ServiceI::start(
                 properties->setProperty(name + ".Node.ThreadPool.Size", os.str());
                 properties->setProperty(name + ".Node.ThreadPool.SizeWarn", "0");
             }
+            if(properties->getProperty(name + ".Node.MessageSizeMax").empty())
+            {
+                properties->setProperty(name + ".Node.MessageSizeMax", 0); // No limit on data exchanged internally
+            }
             Ice::ObjectAdapterPtr nodeAdapter = communicator->createObjectAdapter(name + ".Node");
 
             _instance = new Instance(instanceName, name, communicator, publishAdapter, topicAdapter,
diff --git a/cpp/test/Ice/exceptions/AllTests.cpp b/cpp/test/Ice/exceptions/AllTests.cpp
index f03bdab60b5..13845097f8b 100644
--- a/cpp/test/Ice/exceptions/AllTests.cpp
+++ b/cpp/test/Ice/exceptions/AllTests.cpp
@@ -932,10 +932,30 @@ allTests(const Ice::CommunicatorPtr& communicator)
         catch(const Ice::ConnectionLostException&)
         {
         }
-        catch(...)
+        catch(const Ice::LocalException& ex)
+        {
+            cerr << ex << endl;
+            test(false);
+        }
+        
+        ThrowerPrx thrower2 = ThrowerPrx::uncheckedCast(communicator->stringToProxy("thrower:default -p 12011"));
+        try
+        {
+            thrower2->throwMemoryLimitException(Ice::ByteSeq(2 * 1024 * 1024)); // 2MB (no limits)
+        }
+        catch(const Ice::MemoryLimitException&)
+        {
+        }
+        ThrowerPrx thrower3 = ThrowerPrx::uncheckedCast(communicator->stringToProxy("thrower:default -p 12012"));
+        try
         {
+            thrower3->throwMemoryLimitException(Ice::ByteSeq(1024)); // 1KB limit
             test(false);
         }
+        catch(const Ice::ConnectionLostException&)
+        {
+        }
+
         cout << "ok" << endl;
     }
 
diff --git a/cpp/test/Ice/exceptions/Server.cpp b/cpp/test/Ice/exceptions/Server.cpp
index d50ffc78dd6..32706ccaf8b 100644
--- a/cpp/test/Ice/exceptions/Server.cpp
+++ b/cpp/test/Ice/exceptions/Server.cpp
@@ -19,9 +19,15 @@ int
 run(int, char**, const Ice::CommunicatorPtr& communicator)
 {
     Ice::ObjectAdapterPtr adapter = communicator->createObjectAdapter("TestAdapter");
+    Ice::ObjectAdapterPtr adapter2 = communicator->createObjectAdapter("TestAdapter2");
+    Ice::ObjectAdapterPtr adapter3 = communicator->createObjectAdapter("TestAdapter3");
     Ice::ObjectPtr object = new ThrowerI();
     adapter->add(object, communicator->stringToIdentity("thrower"));
+    adapter2->add(object, communicator->stringToIdentity("thrower"));
+    adapter3->add(object, communicator->stringToIdentity("thrower"));
     adapter->activate();
+    adapter2->activate();
+    adapter3->activate();
     TEST_READY
     communicator->waitForShutdown();
     return EXIT_SUCCESS;
@@ -41,6 +47,10 @@ main(int argc, char* argv[])
         initData.properties->setProperty("Ice.Warn.Connections", "0");
         initData.properties->setProperty("TestAdapter.Endpoints", "default -p 12010:udp");
         initData.properties->setProperty("Ice.MessageSizeMax", "10"); // 10KB max
+        initData.properties->setProperty("TestAdapter2.Endpoints", "default -p 12011");
+        initData.properties->setProperty("TestAdapter2.MessageSizeMax", "0");
+        initData.properties->setProperty("TestAdapter3.Endpoints", "default -p 12012");
+        initData.properties->setProperty("TestAdapter3.MessageSizeMax", "1");
         communicator = Ice::initialize(argc, argv, initData);
         status = run(argc, argv, communicator);
     }
diff --git a/cpp/test/Ice/exceptions/ServerAMD.cpp b/cpp/test/Ice/exceptions/ServerAMD.cpp
index 04f4d5805bc..75de52d1efb 100644
--- a/cpp/test/Ice/exceptions/ServerAMD.cpp
+++ b/cpp/test/Ice/exceptions/ServerAMD.cpp
@@ -19,9 +19,15 @@ int
 run(int, char**, const Ice::CommunicatorPtr& communicator)
 {
     Ice::ObjectAdapterPtr adapter = communicator->createObjectAdapter("TestAdapter");
+    Ice::ObjectAdapterPtr adapter2 = communicator->createObjectAdapter("TestAdapter2");
+    Ice::ObjectAdapterPtr adapter3 = communicator->createObjectAdapter("TestAdapter3");
     Ice::ObjectPtr object = new ThrowerI();
     adapter->add(object, communicator->stringToIdentity("thrower"));
+    adapter2->add(object, communicator->stringToIdentity("thrower"));
+    adapter3->add(object, communicator->stringToIdentity("thrower"));
     adapter->activate();
+    adapter2->activate();
+    adapter3->activate();
     TEST_READY
     communicator->waitForShutdown();
     return EXIT_SUCCESS;
@@ -41,6 +47,10 @@ main(int argc, char* argv[])
         initData.properties->setProperty("Ice.Warn.Connections", "0");
         initData.properties->setProperty("TestAdapter.Endpoints", "default -p 12010:udp");
         initData.properties->setProperty("Ice.MessageSizeMax", "10"); // 10KB max
+        initData.properties->setProperty("TestAdapter2.Endpoints", "default -p 12011");
+        initData.properties->setProperty("TestAdapter2.MessageSizeMax", "0");
+        initData.properties->setProperty("TestAdapter3.Endpoints", "default -p 12012");
+        initData.properties->setProperty("TestAdapter3.MessageSizeMax", "1");
         communicator = Ice::initialize(argc, argv, initData);
         status = run(argc, argv, communicator);
     }