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.

4 files changed, 44 insertions, 17 deletions

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