asynchronously resolve a name to multiple address objects

src/twisted/internet/_resolver.py

@@ -0,0 +1,263 @@
+# -*- test-case-name: twisted.internet.test.test_resolver -*-
+# Copyright (c) Twisted Matrix Laboratories.
+# See LICENSE for details.
+
+"""
+IPv6-aware hostname resolution.
+
+@see: L{IHostnameResolver}
+"""
+
+from __future__ import division, absolute_import
+
+__metaclass__ = type
+
+from socket import (getaddrinfo, AF_INET, AF_INET6, AF_UNSPEC, SOCK_STREAM,
+                    SOCK_DGRAM, gaierror)
+
+from zope.interface import implementer
+
+from twisted.internet.interfaces import (IHostnameResolver, IHostResolution,
+                                         IResolverSimple, IResolutionReceiver)
+from twisted.internet.error import DNSLookupError
+from twisted.internet.defer import Deferred
+from twisted.internet.threads import deferToThreadPool
+from twisted.internet.address import IPv4Address, IPv6Address
+from twisted.logger import Logger
+
+
+@implementer(IHostResolution)
+class HostResolution(object):
+    """
+    The in-progress resolution of a given hostname.
+    """
+
+    def __init__(self, name):
+        """
+        Create a L{HostResolution} with the given name.
+        """
+        self.name = name
+
+
+
+_any = frozenset([IPv4Address, IPv6Address])
+
+_typesToAF = {
+    frozenset([IPv4Address]): AF_INET,
+    frozenset([IPv6Address]): AF_INET6,
+    _any: AF_UNSPEC,
+}
+
+_afToType = {
+    AF_INET: IPv4Address,
+    AF_INET6: IPv6Address,
+}
+
+_transportToSocket = {
+    'TCP': SOCK_STREAM,
+    'UDP': SOCK_DGRAM,
+}
+
+_socktypeToType = {
+    SOCK_STREAM: 'TCP',
+    SOCK_DGRAM: 'UDP',
+}
+
+
+
+@implementer(IHostnameResolver)
+class GAIResolver(object):
+    """
+    L{IHostnameResolver} implementation that resolves hostnames by calling
+    L{getaddrinfo} in a thread.
+    """
+
+    def __init__(self, reactor, getThreadPool=None, getaddrinfo=getaddrinfo):
+        """
+        Create a L{GAIResolver}.
+
+        @param reactor: the reactor to schedule result-delivery on
+        @type reactor: L{IReactorThreads}
+
+        @param getThreadPool: a function to retrieve the thread pool to use for
+            scheduling name resolutions.  If not supplied, the use the given
+            C{reactor}'s thread pool.
+        @type getThreadPool: 0-argument callable returning a
+            L{twisted.python.threadpool.ThreadPool}
+
+        @param getaddrinfo: a reference to the L{getaddrinfo} to use - mainly
+            parameterized for testing.
+        @type getaddrinfo: callable with the same signature as L{getaddrinfo}
+        """
+        self._reactor = reactor
+        self._getThreadPool = (reactor.getThreadPool if getThreadPool is None
+                               else getThreadPool)
+        self._getaddrinfo = getaddrinfo
+
+
+    def resolveHostName(self, resolutionReceiver, hostName, portNumber=0,
+                        addressTypes=None, transportSemantics='TCP'):
+        """
+        See L{IHostnameResolver.resolveHostName}
+
+        @param resolutionReceiver: see interface
+
+        @param hostName: see interface
+
+        @param portNumber: see interface
+
+        @param addressTypes: see interface
+
+        @param transportSemantics: see interface
+
+        @return: see interface
+        """
+        pool = self._getThreadPool()
+        addressFamily = _typesToAF[_any if addressTypes is None
+                                   else frozenset(addressTypes)]
+        socketType = _transportToSocket[transportSemantics]
+        def get():
+            try:
+                return self._getaddrinfo(hostName, portNumber, addressFamily,
+                                         socketType)
+            except gaierror:
+                return []
+        d = deferToThreadPool(self._reactor, pool, get)
+        resolution = HostResolution(hostName)
+        resolutionReceiver.resolutionBegan(resolution)
+        @d.addCallback
+        def deliverResults(result):
+            for family, socktype, proto, cannoname, sockaddr in result:
+                addrType = _afToType[family]
+                resolutionReceiver.addressResolved(
+                    addrType(_socktypeToType.get(socktype, 'TCP'), *sockaddr)
+                )
+            resolutionReceiver.resolutionComplete()
+        return resolution
+
+
+
+@implementer(IHostnameResolver)
+class SimpleResolverComplexifier(object):
+    """
+    A converter from L{IResolverSimple} to L{IHostnameResolver}.
+    """
+
+    _log = Logger()
+
+    def __init__(self, simpleResolver):
+        """
+        Construct a L{SimpleResolverComplexifier} with an L{IResolverSimple}.
+        """
+        self._simpleResolver = simpleResolver
+
+
+    def resolveHostName(self, resolutionReceiver, hostName, portNumber=0,
+                        addressTypes=None, transportSemantics='TCP'):
+        """
+        See L{IHostnameResolver.resolveHostName}
+
+        @param resolutionReceiver: see interface
+
+        @param hostName: see interface
+
+        @param portNumber: see interface
+
+        @param addressTypes: see interface
+
+        @param transportSemantics: see interface
+
+        @return: see interface
+        """
+        resolution = HostResolution(hostName)
+        resolutionReceiver.resolutionBegan(resolution)
+        onAddress = self._simpleResolver.getHostByName(hostName)
+        def addressReceived(address):
+            resolutionReceiver.addressResolved(IPv4Address('TCP', address, 0))
+        def errorReceived(error):
+            if not error.check(DNSLookupError):
+                self._log.failure("while looking up {name} with {resolver}",
+                                  error, name=hostName,
+                                  resolver=self._simpleResolver)
+        onAddress.addCallbacks(addressReceived, errorReceived)
+        def finish(result):
+            resolutionReceiver.resolutionComplete()
+        onAddress.addCallback(finish)
+        return resolution
+
+
+
+@implementer(IResolutionReceiver)
+class FirstOneWins(object):
+    """
+    An L{IResolutionReceiver} which fires a L{Deferred} with its first result.
+    """
+
+    def __init__(self, deferred):
+        """
+        @param deferred: The L{Deferred} to fire when the first resolution
+            result arrives.
+        """
+        self._deferred = deferred
+        self._resolved = False
+
+
+    def resolutionBegan(self, resolution):
+        """
+        See L{IResolutionReceiver.resolutionBegan}
+
+        @param resolution: See L{IResolutionReceiver.resolutionBegan}
+        """
+        self._resolution = resolution
+
+
+    def addressResolved(self, address):
+        """
+        See L{IResolutionReceiver.addressResolved}
+
+        @param address: See L{IResolutionReceiver.addressResolved}
+        """
+        if self._resolved:
+            return
+        self._resolved = True
+        self._deferred.callback(address.host)
+
+
+    def resolutionComplete(self):
+        """
+        See L{IResolutionReceiver.resolutionComplete}
+        """
+        if self._resolved:
+            return
+        self._deferred.errback(DNSLookupError(self._resolution.name))
+
+
+
+@implementer(IResolverSimple)
+class ComplexResolverSimplifier(object):
+    """
+    A converter from L{IHostnameResolver} to L{IResolverSimple}
+    """
+    def __init__(self, nameResolver):
+        """
+        Create a L{ComplexResolverSimplifier} with an L{IHostnameResolver}.
+
+        @param nameResolver: The L{IHostnameResolver} to use.
+        """
+        self._nameResolver = nameResolver
+
+
+    def getHostByName(self, name, timeouts=()):
+        """
+        See L{IResolverSimple.getHostByName}
+
+        @param name: see L{IResolverSimple.getHostByName}
+
+        @param timeouts: see L{IResolverSimple.getHostByName}
+
+        @return: see L{IResolverSimple.getHostByName}
+        """
+        result = Deferred()
+        self._nameResolver.resolveHostName(FirstOneWins(result), name, 0,
+                                           [IPv4Address])
+        return result

src/twisted/internet/address.py

@@ -79,8 +79,23 @@ class IPv6Address(_IPAddress):
     @ivar host: A string containing a colon-separated, hexadecimal formatted
         IPv6 address; for example, "::1".
     @type host: C{str}
+
+    @ivar flowInfo: the IPv6 flow label.  This can be used by QoS routers to
+        identify flows of traffic; you may generally safely ignore it.
+    @type flowInfo: L{int}
+
+    @ivar scopeID: the IPv6 scope identifier - roughly analagous to what
+        interface traffic destined for this address must be transmitted over.
+    @type scopeID: L{int}
     """
 
+    compareAttributes = ('type', 'host', 'port', 'flowInfo', 'scopeID')
+
+    def __init__(self, type, host, port, flowInfo=0, scopeID=0):
+        super(IPv6Address, self).__init__(type, host, port)
+        self.flowInfo = flowInfo
+        self.scopeID = scopeID
+
 
 
 @implementer(IAddress)

src/twisted/internet/base.py

@@ -21,6 +21,11 @@
 from twisted.internet.interfaces import IResolverSimple, IReactorPluggableResolver
 from twisted.internet.interfaces import IConnector, IDelayedCall
 from twisted.internet import fdesc, main, error, abstract, defer, threads
+from twisted.internet._resolver import (
+    GAIResolver as _GAIResolver,
+    ComplexResolverSimplifier as _ComplexResolverSimplifier,
+    SimpleResolverComplexifier as _SimpleResolverComplexifier,
+)
 from twisted.python import log, failure, reflect
 from twisted.python.compat import unicode, iteritems
 from twisted.python.runtime import seconds as runtimeSeconds, platform
@@ -487,6 +492,7 @@ def __init__(self):
         self._startedBefore = False
         # reactor internal readers, e.g. the waker.
         self._internalReaders = set()
+        self._nameResolver = None
         self.waker = None
 
         # Arrange for the running attribute to change to True at the right time
@@ -509,12 +515,45 @@ def installWaker(self):
         raise NotImplementedError(
             reflect.qual(self.__class__) + " did not implement installWaker")
 
+
     def installResolver(self, resolver):
+        """
+        See L{IReactorPluggableResolver}.
+
+        @param resolver: see L{IReactorPluggableResolver}.
+
+        @return: see L{IReactorPluggableResolver}.
+        """
         assert IResolverSimple.providedBy(resolver)
         oldResolver = self.resolver
         self.resolver = resolver
+        self._nameResolver = _SimpleResolverComplexifier(resolver)
         return oldResolver
 
+
+    def installNameResolver(self, resolver):
+        """
+        See L{IReactorPluggableNameResolver}.
+
+        @param resolver: See L{IReactorPluggableNameResolver}.
+
+        @return: see L{IReactorPluggableNameResolver}.
+        """
+        previousNameResolver = self._nameResolver
+        self._nameResolver = resolver
+        self.resolver = _ComplexResolverSimplifier(resolver)
+        return previousNameResolver
+
+
+    @property
+    def nameResolver(self):
+        """
+        Implementation of read-only
+        L{IReactorPluggableNameResolver.nameResolver}.
+        """
+        return self._nameResolver
+
+
     def wakeUp(self):
         """
         Wake up the event loop.
@@ -938,8 +977,9 @@ def argChecker(arg):
         threadpoolShutdownID = None
 
         def _initThreads(self):
+            self.installNameResolver(_GAIResolver(self, self.getThreadPool))
             self.usingThreads = True
-            self.resolver = ThreadedResolver(self)
+
 
         def callFromThread(self, f, *args, **kw):
             """