1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
|
from twisted.internet import reactor
from twisted.internet.stdio import StandardIO
from twisted.internet.protocol import ServerFactory
from twisted.internet.ssl import DefaultOpenSSLContextFactory
from twisted.protocols.basic import LineOnlyReceiver
class Echoer(LineOnlyReceiver):
delimiter = "\n"
def lineReceived(self, line):
self.factory.linecount += 1
self.sendLine("%d: %s" % (self.factory.linecount, line))
class EchoFactory(ServerFactory):
protocol = Echoer
linecount = 0
def main():
factory = EchoFactory()
reactor.listenTCP(12345, factory)
reactor.listenUNIX('echo', factory)
reactor.listenSSL(
12346,
factory,
DefaultOpenSSLContextFactory('server.key', 'server.pem'))
echoer = Echoer()
echoer.factory = factory
StandardIO(echoer)
reactor.run()
def test():
from StringIO import StringIO
from twisted.internet.protocol import FileWrapper
factory = EchoFactory()
firstProtocol = factory.buildProtocol(None)
firstTransport = StringIO()
firstProtocol.makeConnection(FileWrapper(firstTransport))
secondProtocol = factory.buildProtocol(None)
secondTransport = StringIO()
secondProtocol.makeConnection(FileWrapper(secondTransport))
firstProtocol.lineReceived("hello")
assert firstTransport.getvalue() == "1: hello\n"
secondProtocol.lineReceived("world")
assert secondTransport.getvalue() == "2: world\n"
test()
main()
|