ICE-5573 - .NET TimeoutException ignores Dispatcher

- Added test for C++, C# & Java to the dispatcher test
  to ensure that the invocation timeout is throwin
  in the correct thread.
- Pass the connection to the dispatcher invocation in
  more cases.

9 files changed, 75 insertions, 17 deletions

diff --git a/cpp/include/Ice/OutgoingAsync.h b/cpp/include/Ice/OutgoingAsync.h
index 96b5d96bd2d..360ddf18acb 100644
--- a/cpp/include/Ice/OutgoingAsync.h
+++ b/cpp/include/Ice/OutgoingAsync.h
@@ -159,6 +159,7 @@ protected:
     const CommunicatorPtr _communicator;
     const IceInternal::InstancePtr _instance;
     const std::string& _operation;
+    Ice::ConnectionPtr _cachedConnection;
     const IceInternal::CallbackBasePtr _callback;
     const LocalObjectPtr _cookie;
 
diff --git a/cpp/src/Ice/ConnectRequestHandler.cpp b/cpp/src/Ice/ConnectRequestHandler.cpp
index 7ff4609d924..c3b4a4b160d 100644
--- a/cpp/src/Ice/ConnectRequestHandler.cpp
+++ b/cpp/src/Ice/ConnectRequestHandler.cpp
@@ -29,7 +29,8 @@ class FlushRequestsWithException : public DispatchWorkItem
 {
 public:
     
-    FlushRequestsWithException(const ConnectRequestHandlerPtr& handler) : _handler(handler)
+    FlushRequestsWithException(const Ice::ConnectionPtr& connection, const ConnectRequestHandlerPtr& handler) :
+        DispatchWorkItem(connection), _handler(handler)
     {
     }
     
@@ -48,7 +49,8 @@ class FlushSentRequests : public DispatchWorkItem
 {
 public:
     
-    FlushSentRequests(const vector<OutgoingAsyncMessageCallbackPtr>& callbacks) : _callbacks(callbacks)
+    FlushSentRequests(const Ice::ConnectionPtr& connection, const vector<OutgoingAsyncMessageCallbackPtr>& callbacks) :
+        DispatchWorkItem(connection), _callbacks(callbacks)
     {
     }
 
@@ -367,7 +369,7 @@ ConnectRequestHandler::setException(const Ice::LocalException& ex)
     //
     if(!_requests.empty())
     {
-        _reference->getInstance()->clientThreadPool()->execute(new FlushRequestsWithException(this));
+        _reference->getInstance()->clientThreadPool()->execute(new FlushRequestsWithException(_connection, this));
     }
 
     notifyAll();
@@ -483,19 +485,19 @@ ConnectRequestHandler::flushRequests()
         Lock sync(*this);
         assert(!_exception.get() && !_requests.empty());
         _exception.reset(ex.get()->ice_clone());
-        _reference->getInstance()->clientThreadPool()->execute(new FlushRequestsWithException(this));
+        _reference->getInstance()->clientThreadPool()->execute(new FlushRequestsWithException(_connection, this));
     }
     catch(const Ice::LocalException& ex)
     {
         Lock sync(*this);
         assert(!_exception.get() && !_requests.empty());
         _exception.reset(ex.ice_clone());
-        _reference->getInstance()->clientThreadPool()->execute(new FlushRequestsWithException(this));
+        _reference->getInstance()->clientThreadPool()->execute(new FlushRequestsWithException(_connection, this));
     }
 
     if(!sentCallbacks.empty())
     {
-        _reference->getInstance()->clientThreadPool()->execute(new FlushSentRequests(sentCallbacks));
+        _reference->getInstance()->clientThreadPool()->execute(new FlushSentRequests(_connection, sentCallbacks));
     }
         
     //
diff --git a/cpp/src/Ice/OutgoingAsync.cpp b/cpp/src/Ice/OutgoingAsync.cpp
index 3fb2c8ed08b..8b12805ba4f 100644
--- a/cpp/src/Ice/OutgoingAsync.cpp
+++ b/cpp/src/Ice/OutgoingAsync.cpp
@@ -51,8 +51,8 @@ class AsynchronousException : public DispatchWorkItem
 {
 public:
 
-    AsynchronousException(const Ice::AsyncResultPtr& result, const Ice::Exception& ex) :
-        _result(result), _exception(ex.ice_clone())
+    AsynchronousException(const Ice::ConnectionPtr& connection, const Ice::AsyncResultPtr& result, const Ice::Exception& ex) :
+        DispatchWorkItem(connection), _result(result), _exception(ex.ice_clone())
     {
     }
 
@@ -72,7 +72,8 @@ class AsynchronousSent : public DispatchWorkItem
 {
 public:
 
-    AsynchronousSent(const Ice::AsyncResultPtr& result) : _result(result)
+    AsynchronousSent(const Ice::ConnectionPtr& connection, const Ice::AsyncResultPtr& result) :
+        DispatchWorkItem(connection), _result(result)
     {
     }
 
@@ -91,8 +92,8 @@ class AsynchronousTimeout : public DispatchWorkItem
 {
 public:
 
-    AsynchronousTimeout(const IceInternal::RequestHandlerPtr& handler, const Ice::AsyncResultPtr& result) :
-        _handler(handler), _outAsync(OutgoingAsyncMessageCallbackPtr::dynamicCast(result))
+    AsynchronousTimeout(const Ice::ConnectionPtr& connection, const IceInternal::RequestHandlerPtr& handler, const Ice::AsyncResultPtr& result) :
+        DispatchWorkItem(connection), _handler(handler), _outAsync(OutgoingAsyncMessageCallbackPtr::dynamicCast(result))
     {
         assert(_outAsync);
     }
@@ -268,7 +269,7 @@ Ice::AsyncResult::__invokeSentAsync()
     //
     try
     {
-        _instance->clientThreadPool()->execute(new AsynchronousSent(this));
+        _instance->clientThreadPool()->execute(new AsynchronousSent(_cachedConnection, this));
     }
     catch(const Ice::CommunicatorDestroyedException&)
     {
@@ -300,7 +301,7 @@ Ice::AsyncResult::__invokeExceptionAsync(const Ice::Exception& ex)
     // CommunicatorDestroyedException is the only exception that can propagate directly
     // from this method.
     //
-    _instance->clientThreadPool()->execute(new AsynchronousException(this, ex));
+    _instance->clientThreadPool()->execute(new AsynchronousException(_cachedConnection, this, ex));
 }
 
 void
@@ -342,7 +343,16 @@ Ice::AsyncResult::runTimerTask() // Implementation of TimerTask::runTimerTask()
 
     if(handler)
     {
-        _instance->clientThreadPool()->execute(new AsynchronousTimeout(handler, this));
+        Ice::ConnectionPtr connection;
+        try
+        {
+            connection = handler->getConnection(false);
+        }
+        catch(const Ice::LocalException&)
+        {
+            // Ignore.
+        }
+        _instance->clientThreadPool()->execute(new AsynchronousTimeout(connection, handler, this));
     }
 }
 
@@ -507,6 +517,7 @@ IceInternal::OutgoingAsync::__prepare(const std::string& operation, OperationMod
 AsyncStatus
 IceInternal::OutgoingAsync::__send(const Ice::ConnectionIPtr& connection, bool compress, bool response)
 {
+    _cachedConnection = connection;
     return connection->sendAsyncRequest(this, compress, response);
 }
 
@@ -838,6 +849,7 @@ IceInternal::BatchOutgoingAsync::BatchOutgoingAsync(const CommunicatorPtr& commu
 AsyncStatus 
 IceInternal::BatchOutgoingAsync::__send(const Ice::ConnectionIPtr& connection, bool, bool)
 {
+    _cachedConnection = connection;
     return connection->flushAsyncBatchRequests(this);
 }
 
diff --git a/cpp/src/Ice/ThreadPool.cpp b/cpp/src/Ice/ThreadPool.cpp
index d7884f54964..d7c26aced63 100644
--- a/cpp/src/Ice/ThreadPool.cpp
+++ b/cpp/src/Ice/ThreadPool.cpp
@@ -141,6 +141,14 @@ class ThreadPoolDestroyedException
 
 }
 
+
+IceInternal::DispatchWorkItem::DispatchWorkItem() {
+}
+
+IceInternal::DispatchWorkItem::DispatchWorkItem(const Ice::ConnectionPtr& connection) : _connection(connection) {
+    
+}
+
 void
 IceInternal::DispatchWorkItem::execute(ThreadPoolCurrent& current)
 {
@@ -150,7 +158,7 @@ IceInternal::DispatchWorkItem::execute(ThreadPoolCurrent& current)
     {
         try
         {
-            dispatcher->dispatch(this, 0);
+            dispatcher->dispatch(this, _connection);
         }
         catch(const std::exception& ex)
         {
diff --git a/cpp/src/Ice/ThreadPool.h b/cpp/src/Ice/ThreadPool.h
index 6d2c9d283e5..f4df601605c 100644
--- a/cpp/src/Ice/ThreadPool.h
+++ b/cpp/src/Ice/ThreadPool.h
@@ -78,7 +78,6 @@ public:
     }
     void finish(const EventHandlerPtr&);
     void execute(const ThreadPoolWorkItemPtr&);
-
     void joinWithAllThreads();
 
     std::string prefix() const;
@@ -184,9 +183,15 @@ public:
 
 class DispatchWorkItem : public ThreadPoolWorkItem, public Ice::DispatcherCall
 {
-private:
+public:
+
+    DispatchWorkItem();
+    DispatchWorkItem(const Ice::ConnectionPtr& connection);
  
+ private:
+
     virtual void execute(ThreadPoolCurrent&);
+    Ice::ConnectionPtr _connection;
 };
 
 class ThreadPoolWorkQueue : public EventHandler, public IceUtil::Mutex
diff --git a/cpp/test/Ice/dispatcher/AllTests.cpp b/cpp/test/Ice/dispatcher/AllTests.cpp
index 7e82afcf37d..a4a7cc25c5c 100644
--- a/cpp/test/Ice/dispatcher/AllTests.cpp
+++ b/cpp/test/Ice/dispatcher/AllTests.cpp
@@ -52,6 +52,18 @@ public:
         called();
     }
 
+    void responseEx()
+    {
+        test(false);
+    }
+
+    void exceptionEx(const ::Ice::Exception& ex)
+    {
+        test(dynamic_cast<const Ice::InvocationTimeoutException*>(&ex));
+        test(Dispatcher::isDispatcherThread());
+        called();
+    }
+
     void
     payload()
     {
@@ -123,6 +135,15 @@ allTests(const Ice::CommunicatorPtr& communicator)
         i->begin_op(callback);
         cb->check();
 
+        {
+            //
+            // Expect InvocationTimeoutException.
+            //
+            Test::TestIntfPrx to = p->ice_invocationTimeout(250);
+            to->begin_sleep(500, Test::newCallback_TestIntf_sleep(cb, &Callback::responseEx, &Callback::exceptionEx));
+            cb->check();
+        }
+
         testController->holdAdapter();
 
         Test::Callback_TestIntf_opWithPayloadPtr callback2 =
diff --git a/cpp/test/Ice/dispatcher/Test.ice b/cpp/test/Ice/dispatcher/Test.ice
index 857acae24a9..0379b4c7f4c 100644
--- a/cpp/test/Ice/dispatcher/Test.ice
+++ b/cpp/test/Ice/dispatcher/Test.ice
@@ -18,6 +18,7 @@ module Test
 interface TestIntf
 {
     void op();
+    void sleep(int to);
     void opWithPayload(Ice::ByteSeq seq);
     void shutdown();
 };
diff --git a/cpp/test/Ice/dispatcher/TestI.cpp b/cpp/test/Ice/dispatcher/TestI.cpp
index 2a56d659f82..9140fc55fb0 100644
--- a/cpp/test/Ice/dispatcher/TestI.cpp
+++ b/cpp/test/Ice/dispatcher/TestI.cpp
@@ -11,6 +11,7 @@
 #include <Ice/Ice.h>
 #include <Dispatcher.h>
 #include <TestCommon.h>
+#include <IceUtil/Thread.h>
 
 using namespace std;
 
@@ -21,6 +22,12 @@ TestIntfI::op(const Ice::Current&)
 }
 
 void
+TestIntfI::sleep(Ice::Int to, const Ice::Current&)
+{
+    IceUtil::ThreadControl::sleep(IceUtil::Time::milliSeconds(to));
+}
+
+void
 TestIntfI::opWithPayload(const Ice::ByteSeq&, const Ice::Current&)
 {
     test(Dispatcher::isDispatcherThread());
diff --git a/cpp/test/Ice/dispatcher/TestI.h b/cpp/test/Ice/dispatcher/TestI.h
index 99c64dd20d1..228a2842753 100644
--- a/cpp/test/Ice/dispatcher/TestI.h
+++ b/cpp/test/Ice/dispatcher/TestI.h
@@ -20,6 +20,7 @@ class TestIntfI : virtual public Test::TestIntf
 public:
 
     virtual void op(const Ice::Current&);
+    virtual void sleep(Ice::Int, const Ice::Current&);
     virtual void opWithPayload(const Ice::ByteSeq&, const Ice::Current&);
     virtual void shutdown(const Ice::Current&);
 };