Ticket #3290: test.py

File test.py, 0.7 KB (added by kvogt, 6 years ago)

Test file for demonstrating reactor pausing

Line 
1from twisted.internet import protocol, reactor
2from twisted.application import service, internet
3
4class NotifyProtocol(protocol.Protocol):
5    def __init__(self):
6        self.host_key = None
7    def dataReceived(self, data):
8        self.factory.count += 1
9        print "%s clients have connected." % self.factory.count
10        self.transport.write("HTTP/1.x 200 OK\r\n\r\n")
11
12class NotifyFactory(protocol.ServerFactory):
13    protocol = NotifyProtocol
14    def __init__(self):
15        self.count = 0
16
17application = service.Application('NotifyServer')
18internet.TCPServer(2000, NotifyFactory(), interface="0.0.0.0", backlog=50000).setServiceParent(service.IServiceCollection(application))