- Added support for non-blocking AMI/batch requests, connection

  creation.
- Added support for AMI oneway requests.
- Changed collocation optimization to not perform any DNS lookups.

1 files changed, 104 insertions, 1 deletions

diff --git a/cpp/src/Ice/EndpointI.cpp b/cpp/src/Ice/EndpointI.cpp
index 45f88a6cde8..f3505005f28 100644
--- a/cpp/src/Ice/EndpointI.cpp
+++ b/cpp/src/Ice/EndpointI.cpp
@@ -8,7 +8,110 @@
 // **********************************************************************
 
 #include <Ice/EndpointI.h>
+#include <Ice/Instance.h>
+#include <Ice/LocalException.h>
+#include <Ice/Network.h>
 
-using namespace Ice;
+using namespace std;
+using namespace IceInternal;
 
 Ice::LocalObject* IceInternal::upCast(EndpointI* p) { return p; }
+IceUtil::Shared* IceInternal::upCast(EndpointHostResolver* p) { return p; }
+
+vector<ConnectorPtr>
+IceInternal::EndpointI::connectors(const vector<struct sockaddr_in>& addrs) const
+{
+    //
+    // This method must be extended by endpoints which use the EndpointHostResolver to create
+    // connectors from IP addresses.
+    //
+    assert(false);
+    return vector<ConnectorPtr>();
+}
+
+IceInternal::EndpointHostResolver::EndpointHostResolver(const InstancePtr& instance) :
+    _instance(instance),
+    _destroyed(false)
+{
+    start();
+}
+
+void
+IceInternal::EndpointHostResolver::resolve(const string& host, int port, const EndpointIPtr& endpoint, 
+                                           const EndpointI_connectorsPtr& callback)
+{ 
+    //
+    // Try to get the addresses without DNS lookup. If this doesn't work, we queue a resolve
+    // entry and the thread will take care of getting the endpoint addresses.
+    //
+    vector<struct sockaddr_in> addrs = getAddresses(host, port, false);
+    if(!addrs.empty())
+    {
+        callback->connectors(endpoint->connectors(addrs));
+        return;
+    }
+
+    Lock sync(*this);
+    assert(!_destroyed);
+
+    ResolveEntry entry;
+    entry.host = host;
+    entry.port = port;
+    entry.endpoint = endpoint;
+    entry.callback = callback;
+    _queue.push_back(entry);
+    notify();
+}
+
+void
+IceInternal::EndpointHostResolver::destroy()
+{
+    Lock sync(*this);
+    assert(!_destroyed);
+    _destroyed = true;
+    notify();
+}
+
+void
+IceInternal::EndpointHostResolver::run()
+{
+    if(_instance->initializationData().threadHook)
+    {
+        _instance->initializationData().threadHook->start();
+    }
+
+    while(true)
+    {
+        ResolveEntry resolve;
+
+        {
+            Lock sync(*this);
+
+            while(!_destroyed && _queue.empty())
+            {
+                wait();
+            }
+
+            if(_destroyed)
+            {
+                break;
+            }
+
+            resolve = _queue.front();
+            _queue.pop_front();
+        }
+
+        resolve.callback->connectors(resolve.endpoint->connectors(getAddresses(resolve.host, resolve.port)));
+    }
+
+    for(deque<ResolveEntry>::const_iterator p = _queue.begin(); p != _queue.end(); ++p)
+    {
+        p->callback->exception(Ice::CommunicatorDestroyedException(__FILE__, __LINE__));
+    }
+    _queue.clear();
+
+    if(_instance->initializationData().threadHook)
+    {
+        _instance->initializationData().threadHook->stop();
+    }
+}