Index: twisted/test/test_endpoints.py =================================================================== --- twisted/test/test_endpoints.py (revision 0) +++ twisted/test/test_endpoints.py (revision 0) @@ -0,0 +1,166 @@ +from twisted.trial import unittest +from twisted.internet import (defer, error, interfaces, reactor, + _sslverify as sslverify) +from twisted.internet.address import IPv4Address, UNIXAddress +from twisted.internet.protocol import ClientFactory, Protocol, ServerFactory +from twisted.internet.endpoints import (TCPEndpoint, UNIXEndpoint) +from twisted.test.test_sslverify import makeCertificate + +class ServerProtocol(Protocol): + def connectionMade(self): + self.factory.onConnectionMade.callback(self) + + def connectionLost(self, *a): + self.factory.onConnectionLost.callback(self) + +class ClientProtocol(Protocol): + def connectionMade(self): + self.factory.onConnectionMade.callback(self) + + def connectionLost(self, *a): + self.factory.onConnectionLost.callback(self) + +class MyServerFactory(ServerFactory): + protocol = ServerProtocol + + def __init__(self): + self.onConnectionMade = defer.Deferred() + self.onConnectionLost = defer.Deferred() + +class MyClientFactory(ClientFactory): + protocol = ClientProtocol + + def __init__(self): + self.onConnectionMade = defer.Deferred() + self.onConnectionLost = defer.Deferred() + +class PortAndConnectorCleanerUpper(unittest.TestCase): + def setUp(self): + self.listeningPorts = [] + self.clientConnections = [] + + def tearDown(self): + map(lambda p: p.stopListening(), self.listeningPorts) + map(lambda c: c.disconnect(), self.clientConnections) + +class EndpointTestCaseMixin(object): + def test_EndpointConnectSuccess(self): + """Test that Endpoint can connect and returns a deferred who + gets called back with a protocol instance. + """ + sf = MyServerFactory() + p = self.createServer(sf) + addr = p.getHost() + ep = self.createEndpoint(addr) + cf = MyClientFactory() + d = ep.connect(reactor, cf) + self.assertTrue(isinstance(d, defer.Deferred)) + def onConnectSuccess(proto): + self.assertTrue(interfaces.IProtocol.providedBy(proto)) + proto.transport.loseConnection() + d.addCallback(onConnectSuccess) + return defer.gatherResults([sf.onConnectionMade, cf.onConnectionMade, d]) + + def test_EndpointConnectFailure(self): + """Test that if an Endpoint tries to connect to a none + listening port that it gets a ConnectError failure. + """ + p = self.createServer(MyServerFactory()) + addr = p.getHost() + p.loseConnection() + + ep = self.createEndpoint(addr) + d = ep.connect(reactor, MyClientFactory()) + self.failUnlessFailure(d, error.ConnectError) + return d + + def test_EndpointListenSuccess(self): + """Test that Endpoint can listen and returns a deferred that + gets called back with a port instance. + """ + ep = self.createEndpoint() + sf = MyServerFactory() + d = ep.listen(reactor, sf) + self.assertTrue(isinstance(d, defer.Deferred)) + def onListenSuccess(port): + self.assertTrue(interfaces.IListeningPort.providedBy(port)) + self.listeningPorts.append(port) + return port.getHost() + d.addCallback(onListenSuccess) + def connectTo(addr): + self.createClient(addr, MyClientFactory()) + d.addCallback(connectTo) + return defer.gatherResults([sf.onConnectionMade, d]) + + def test_EndpointListenFailure(self): + """Test that if Endpoint tries to listen on an already listening + port, that a CannotListenError failure is errbacked. + """ + p = self.createServer(MyServerFactory()) + addr = p.getHost() + ep = self.createEndpoint(addr) + d = ep.listen(reactor, MyServerFactory()) + self.failUnlessFailure(d, error.CannotListenError) + return d + +class TCPEndpointsTestCase(PortAndConnectorCleanerUpper, EndpointTestCaseMixin): + def createServer(self, factory): + p = reactor.listenTCP(0, factory) + self.listeningPorts.append(p) + return p + + def createClient(self, address, factory): + c = reactor.connectTCP(address.host, address.port, factory) + self.clientConnections.append(c) + return c + + def createEndpoint(self, address=None): + if not address: + address = IPv4Address("TCP", "localhost", 0) + return TCPEndpoint(address.host, address.port) + +class SSLEndpointsTestCase(PortAndConnectorCleanerUpper, EndpointTestCaseMixin): + + def setUpClass(self): + self.sKey, self.sCert = makeCertificate( + O="Server Test Certificate", + CN="server") + self.cKey, self.cCert = makeCertificate( + O="Client Test Certificate", + CN="client") + self.serverSSLContext = sslverify.OpenSSLCertificateOptions(privateKey=self.sKey, certificate=self.sCert, requireCertificate=False) + self.clientSSLContext = sslverify.OpenSSLCertificateOptions(requireCertificate=False) + + def createServer(self, factory): + p = reactor.listenSSL(0, factory, self.serverSSLContext) + self.listeningPorts.append(p) + return p + + def createClient(self, address, factory): + c = reactor.connectSSL(address.host, address.port, factory, + self.clientSSLContext) + self.clientConnections.append(c) + return c + + def createEndpoint(self, address=None): + if not address: + address = IPv4Address("TCP", "localhost", 0) + return TCPEndpoint(address.host, address.port, + sslContextFactory=self.clientSSLContext) + + +class UNIXEndpointsTestCase(PortAndConnectorCleanerUpper, EndpointTestCaseMixin): + def createServer(self, factory): + p = reactor.listenUNIX(self.mktemp(), factory) + self.listeningPorts.append(p) + return p + + def createClient(self, address, factory): + c = reactor.connectUNIX(address.name, factory) + self.clientConnections.append(c) + return c + + def createEndpoint(self, address=None): + if not address: + address = UNIXAddress(self.mktemp()) + return UNIXEndpoint(address.name) Index: twisted/internet/endpoints.py =================================================================== --- twisted/internet/endpoints.py (revision 0) +++ twisted/internet/endpoints.py (revision 0) @@ -0,0 +1,132 @@ +# -*- test-case-name: twisted.test.test_endpoints -*- + +from zope.interface import implements, providedBy, directlyProvides + +from twisted.internet import interfaces +from twisted.internet import defer, protocol +from twisted.internet.protocol import ClientFactory, Protocol + +class _WrappingProtocol(Protocol): + """I wrap another protocol in order to notify my user when a connection has + been made. + """ + def __init__(self, factory, wrappedProtocol): + self.factory = factory + self.wrappedProtocol = wrappedProtocol + + def connectionMade(self): + """XXX: As soon as I am connected, I connect my wrappedProtocol, giving + it my transport. Is it okay for a transport to be associated with more + than one protocol? Transport calls dataReceived on me and I in turn call + dataReceived on my wrappedProtocol. The wrappedProtocol may call + transport.write or transport.loseConnection etc + """ + + self.wrappedProtocol.makeConnection(self.transport) + self.factory.deferred.callback(self.wrappedProtocol) + + def dataReceived(self, data): + return self.wrappedProtocol.dataReceived(data) + + def connectionLost(self, reason): + return self.wrappedProtocol.connectionLost(reason) + +class _WrappingFactory(ClientFactory): + protocol = _WrappingProtocol + + def __init__(self, wrappedFactory): + self.wrappedFactory = wrappedFactory + self.deferred = defer.Deferred() + + def buildProtocol(self, addr): + try: + proto = self.wrappedFactory.buildProtocol(addr) + except: + self.deferred.errback() + else: + return self.protocol(self, proto) + + def clientConnectionFailed(self, connector, reason): + self.deferred.errback(reason) + + +class TCPEndpoint(object): + implements(interfaces.IClientEndpoint, interfaces.IServerEndpoint) + + def __init__(self, host, port, connectArgs={}, listenArgs={}, + sslContextFactory=None): + """ + @param host: A hostname, used only when connecting + @param port: The port number, used both when connecting and listening + @param connectArgs: An optional dict of keyword args that will be passed + to L{twisted.internet.interfaces.IReactorTCP.connectTCP} + @param listenArgs: An optional dict of keyword args that will be passed + to L{twisted.internet.interfaces.IReactorTCP.listenTCP} + @param sslContextFactory: An optional instance of + L{twisted.internet._sslverify.OpenSSLCertificateOptions}. If given, it + makes L{connect} and L{listen} use the corresponding methods from + L{twisted.internet.interfaces.IReactorSSL} + """ + self.host = host + self.port = port + self.connectArgs = dict(timeout=30, bindAddress=None) + self.connectArgs.update(connectArgs) + self.listenArgs = dict(backlog=50, interface='') + self.listenArgs.update(listenArgs) + self.sslContextFactory = sslContextFactory + + def connect(self, reactor, clientFactory): + wf = _WrappingFactory(clientFactory) + connectArgs = self.connectArgs + connectMethod = reactor.connectTCP + if self.sslContextFactory: + connectMethod = reactor.connectSSL + connectArgs["contextFactory"] = self.sslContextFactory + + d = defer.execute(connectMethod, self.host, self.port, wf, **connectArgs) + + d.addCallback(lambda _: wf.deferred) + + return d + + def listen(self, reactor, serverFactory): + wf = _WrappingFactory(serverFactory) + listenArgs = self.listenArgs + listenMethod = reactor.listenTCP + if self.sslContextFactory: + listenMethod = reactor.listenSSL + listenArgs["contextFactory"] = self.sslContextFactory + return defer.execute(listenMethod, self.port, wf, **listenArgs) + +class UNIXEndpoint(object): + implements(interfaces.IClientEndpoint, interfaces.IServerEndpoint) + + def __init__(self, address, connectArgs={}, listenArgs={}): + """ + @param address: The path to the Unix socket file, used both when + connecting and listening + @param connectArgs: A dict of keyword args that will be passed to + L{twisted.internet.interfaces.IReactorUNIX.connectUNIX} + @param listenArgs: A dict of keyword args that will be passed to + L{twisted.internet.interfaces.IReactorUNIX.listenUNIX} + """ + + self.address = address + self.connectArgs = dict(timeout=30, checkPID=0) + self.connectArgs.update(connectArgs) + self.listenArgs = dict(backlog=50, mode=0666, wantPID=0) + self.listenArgs.update(listenArgs) + + def connect(self, reactor, clientFactory): + wf = _WrappingFactory(clientFactory) + d = defer.execute(reactor.connectUNIX, self.address, wf, + **self.connectArgs) + + d.addCallback(lambda _: wf.deferred) + + return d + + def listen(self, reactor, serverFactory): + wf = _WrappingFactory(serverFactory) + return defer.execute(reactor.listenUNIX, self.address, wf, + **self.listenArgs) Index: twisted/internet/interfaces.py =================================================================== --- twisted/internet/interfaces.py (revision 18161) +++ twisted/internet/interfaces.py (working copy) @@ -18,6 +18,11 @@ Default implementations are in L{twisted.internet.address}. """ + + def buildEndpoint(): + """ + @return: an instance providing both L{IClientEndpoint} and L{IServerEndpoint} + """ ### Reactor Interfaces @@ -1294,3 +1299,26 @@ def leaveGroup(addr, interface=""): """Leave multicast group, return Deferred of success.""" + +class IClientEndpoint(Interface): + """Object that represents a remote endpoint that we wish to connect to. + """ + def connect(reactor, clientFactory): + """ + @param reactor: The reactor + @param clientFactory: A provider of L{IProtocolFactory} + @return: A L{Deferred} that results in an L{IProtocol} upon successful + connection otherwise a L{ConnectError} + """ + +class IServerEndpoint(Interface): + """Object representing an endpoint where we will listen for connections. + """ + + def listen(callable): + """ + @param reactor: The reactor + @param serverFactory: A provider of L{IProtocolFactory} + @return: A L{Deferred} that results in an L{IListeningPort} or an + L{CannotListenError} + """ Index: twisted/internet/address.py =================================================================== --- twisted/internet/address.py (revision 18161) +++ twisted/internet/address.py (working copy) @@ -5,10 +5,12 @@ """Address objects for network connections.""" import warnings, os + from zope.interface import implements + from twisted.internet.interfaces import IAddress +from twisted.internet.endpoints import TCPEndpoint, UNIXEndpoint - class IPv4Address(object): """ Object representing an IPv4 socket endpoint. @@ -34,6 +36,12 @@ self.port = port self._bwHack = _bwHack + def buildEndpoint(self): + if self.type == "TCP": + return TCPEndpoint(self.host, self.port) + else: + raise NotImplementedError + def __getitem__(self, index): warnings.warn("IPv4Address.__getitem__ is deprecated. Use attributes instead.", category=DeprecationWarning, stacklevel=2) @@ -70,7 +78,10 @@ def __init__(self, name, _bwHack='UNIX'): self.name = name self._bwHack = _bwHack - + + def buildEndpoint(self): + return UNIXEndpoint(self.name) + def __getitem__(self, index): warnings.warn("UNIXAddress.__getitem__ is deprecated. Use attributes instead.", category=DeprecationWarning, stacklevel=2)