꼬인 응용 프로그램 테스트 - 클라이언트로드

Question

나는 Twisted 기반 서버를 작성 했으므로 twisted도 사용하여 테스트하고 싶습니다.

하지만 동시에 여러 요청을 시작하여 부하 테스트를 작성하고 싶습니다.

하지만이 문제에 stucked이기 때문에 내가 트위스트, 주로 클라이언트 측의 개념을하지 않았다고 믿는다

from twisted.internet import reactor, protocol 
from threading import Thread 
from twisted.protocols.basic import LineReceiver 

__author__="smota" 
__date__ ="$30/10/2009 17:17:50$" 

class SquitterClient(LineReceiver): 

    def connectionMade(self): 
     self.sendLine("message from " % threading.current_thread().name); 
     pass 

    def connectionLost(self, reason): 
     print "connection lost" 

    def sendMessage(self, msg): 
     for m in [ "a", "b", "c", "d", "e"]: 
      self.sendLine(msg % " - " % m); 

class SquitterClientFactory(protocol.ClientFactory): 
    protocol = SquitterClient 

    def clientConnectionFailed(self, connector, reason): 
     print "Connection failed - goodbye!" 
     reactor.stop() 

    def clientConnectionLost(self, connector, reason): 
     print "Connection lost - goodbye!" 
     reactor.stop() 

def createAndRun(): 
    f = SquitterClientFactory() 
    reactor.connectTCP("localhost", 4010, f) 
    reactor.run(installSignalHandlers=0) 

# this connects the protocol to a server runing on port 8000 
def main(): 
    for n in range(0,10): 
     th=Thread(target=createAndRun) 
     th.start() 

# this only runs if the module was *not* imported 
if __name__ == '__main__': 
    main() 

socket_client.py:35 : DeprecationWarning : 원자로 이미 실행 중! 이 문제는 트위스트 8.0
reactor.run (installSignalHandlers = 0) 내가 놓친 게 무엇

이후 을되지 않습니다?

테스트 방법은 무엇입니까?

은 실패의 직접적인 원인은 당신이 원자로를 여러 번에() 실행을 호출하는이 시도이다,

사무엘

Answer 1

을 주셔서 감사합니다. 한 번만 run()을 호출해야합니다. 나는 당신이 여러 개의 원자로를 가질 것으로 예상하고 있다고 생각하지만, 실제로는 하나의 원자로 만 가지고있다. 나쁜 점은 여러 개의 원자로를 갖는 것이 어렵거나 불가능하다는 것입니다. 좋은 점은 불필요하다는 것입니다. 사실 다중 스레드가 필요하지 않습니다. 하나의 원자로에서 여러 개의 클라이언트 연결을 다중 연결을 수신 할 수있는 정도로 쉽게 멀티 플렉스 할 수 있습니다.

예제 코드를 수정하면 다음과 같은 코드가 작동합니다. 핵심 아이디어는 여러 가지 원자로를 동시에 사용할 필요가 없다는 것입니다. 어쨌든 일반적인 Python 구현과 동시에있을 수있는 유일한 방법은 I/O입니다.

from twisted.internet import reactor, protocol 
from twisted.protocols.basic import LineReceiver 

__author__="smota" 
__date__ ="$30/10/2009 17:17:50$" 

class SquitterClient(LineReceiver): 
    def connectionMade(self): 
     self.messageCount = 0 
     # The factory provides a reference to itself, we'll use it to enumerate the clients 
     self.factory.n += 1 
     self.name = "Client %d" %self.factory.n 

     # Send initial message, and more messages a bit later 
     self.sendLine("Client %s starting!" % self.name); 
     reactor.callLater(0.5, self.sendMessage, "Message %d" %self.messageCount) 

    def connectionLost(self, reason): 
     print "connection lost" 

    def sendMessage(self, msg): 
     for m in [ "a", "b", "c", "d", "e"]: 
      self.sendLine("Copy %s of message %s from client %s!" % (m, msg, self.name)) 
     if self.factory.stop: 
      self.sendLine("Client %s disconnecting!" % self.name) 
      self.transport.loseConnection() 
     else: 
      self.messageCount += 1 
      reactor.callLater(0.5, self.sendMessage, "Message %d" %self.messageCount) 

class SquitterClientFactory(protocol.ClientFactory): 
    protocol = SquitterClient 

    def __init__(self): 
     self.n = 0 
     self.stop = False 

    def stopTest(): 
     self.stop = True 

    def clientConnectionFailed(self, connector, reason): 
     print "Connection failed - goodbye!" 

    def clientConnectionLost(self, connector, reason): 
     print "Connection lost - goodbye!" 

# this connects the protocol to a server running on port 8000 
def main(): 
    # Create 10 clients 

    f = SquitterClientFactory() 
    for i in range(10): 
     reactor.connectTCP("localhost", 8000, f) 

    # Schedule end of test in 10 seconds 
    reactor.callLater(10, f.stopTest) 

    # And let loose the dogs of war 
    reactor.run() 

# this only runs if the module was *not* imported 
if __name__ == '__main__': 
    main()