python twistedを初めて使う
以下のプログラムはすべて『Python.UNIXとLinuxシステム管理ガイド』から来ています.
Tcpポートの検出をtwistedで実現
実行結果:
[root@centos python]# python twisted_check_tcp_port.py 127.0.0.1:80
Testing 127.0.0.1:80
Created a new protocol
Connection made
[root@centos python]# python twisted_check_tcp_port.py 127.0.0.1:8080
Testing 127.0.0.1:8080
Connection failed because [Failure instance: Traceback (failure with no frames):: Connection was refused by other side: 111: Connection refused.
]
twistedでperspective brokerを実現
serverエンド
クライアント側
そのうち
サービス側の実行(正しい場合、サービス側は何も印刷しません)
[root@centos python]# python twisted_perspectiv_broker.py
クライアントの実行(正しく実行すると、リストするディレクトリがリストされ、ディレクトリがない場合は空のリストが返されます)
[root@centos python]# python twisted_perspectiv_broker_client.py
['share', 'libexec', 'games', 'tmp', 'etc', 'lib', 'sbin', 'X11R6', 'kerberos', 'src', 'include', 'local', 'bin']
Tcpポートの検出をtwistedで実現
twisted_check_tcp_port.py
#!/usr/bin/env python
from twisted.internet import reactor, protocol
import sys
class PortCheckerProtocol(protocol.Protocol):
def __init__(self):
print "Created a new protocol"
def connectionMade(self):
print "Connection made"
reactor.stop()
class PortCheckerClientFactory(protocol.ClientFactory):
protocol = PortCheckerProtocol
def clientConnectionFailed(self, connector, reason):
print "Connection failed because", reason
reactor.stop()
if __name__ == '__main__':
host, port = sys.argv[1].split(':')
factory = PortCheckerClientFactory()
print "Testing %s" % sys.argv[1]
reactor.connectTCP(host, int(port), factory)
reactor.run()
実行結果:
[root@centos python]# python twisted_check_tcp_port.py 127.0.0.1:80
Testing 127.0.0.1:80
Created a new protocol
Connection made
[root@centos python]# python twisted_check_tcp_port.py 127.0.0.1:8080
Testing 127.0.0.1:8080
Connection failed because [Failure instance: Traceback (failure with no frames):
]
twistedでperspective brokerを実現
serverエンド
twisted_perspectiv_broker.py
#!/usr/bin/env python
import os
from twisted.spread import pb
from twisted.internet import reactor
class PBDirLister(pb.Root):
def remote_ls(self, directory):
try:
return os.listdir(directory)
except OSError:
return []
def remote_ls_boom(self, directory):
return os.listdir(directory)
if __name__ == '__main__':
reactor.listenTCP(9876, pb.PBServerFactory(PBDirLister()))
reactor.run()
クライアント側
twisted_perspectiv_broker_client.py
#!/usr/bin/python env
from twisted.spread import pb
from twisted.internet import reactor
def handle_err(reason):
print "an error occurred", reason
reactor,stop()
def call_ls(def_call_obj):
return def_call_obj.callRemote('ls', '/usr')
def print_ls(print_result):
print print_result
reactor.stop()
if __name__ == '__main__':
factory = pb.PBClientFactory()
reactor.connectTCP('localhost', 9876, factory)
d = factory.getRootObject()
d.addCallback(call_ls)
d.addCallback(print_ls)
d.addErrback(handle_err)
reactor.run()
そのうち
def_call_obj.callRemote(
'ls'
,
'/usr'
) ls ls_boom
実行結果:サービス側の実行(正しい場合、サービス側は何も印刷しません)
[root@centos python]# python twisted_perspectiv_broker.py
クライアントの実行(正しく実行すると、リストするディレクトリがリストされ、ディレクトリがない場合は空のリストが返されます)
[root@centos python]# python twisted_perspectiv_broker_client.py
['share', 'libexec', 'games', 'tmp', 'etc', 'lib', 'sbin', 'X11R6', 'kerberos', 'src', 'include', 'local', 'bin']