from twisted.internet import reactor,defer,protocol
class CallbackAndDisconnectProtocol(protocol.Protocol): def connectionMade(self): self.factory.deferred.callback("Connected!") self.transport.loseConnection()
class ConnectionTestFactory(protocol.ClientFactory): protocol=CallbackAndDisconnectProtocol def init(self): self.deferred=defer.Deferred() def clientConnectionFailed(self,connector,reason): self.deferred.errback(reason)
def testConnect(host,port): testFactory=ConnectionTestFactory() reactor.connectTCP(host,port,testFactory) return testFactory.deferred def handleSuccess(result,port): print("Connect to port %i"%port) reactor.stop() def handleFailure(failure,port): print("Error connecting to port %i: %s"%(port,failure.getErrorMessage())) reactor.stop()
if name=="main":
import sys
if not len(sys.argv)==3:
print("Usage: connectiontest.py host port")
sys.exit(1)
host=sys.argv[1]
port=int(sys.argv[2])
connecting=testConnect(host,port)
connecting.addCallback(handleSuccess,port)
connecting.addErrback(handleFailure,port)
reactor.run()
def handleAllResules(results,ports): for port,resultInfo in zip(ports,results): success,result=resultInfo if success: print('Connected to port %i' % port) reactor.stop()
import sys host=sys.argv[1] ports=range(1,201) testers=[testConnect(host,port) for port in ports] defer.DeferredList(testers,consumeErrors=True).addCallback(handleAllResults,ports) reactor.run()
