/usr/share/doc/pyro/examples/testserver.py is in pyro-examples 1:3.14-4.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 62 63 64 65 66 67 68 | #
# Test server base implementation, used by a lot of the tests,
# to avoid having to repeat this boilerplate code time after time.
#
import sys
import Pyro.naming
import Pyro.core
from Pyro.protocol import getHostname
from Pyro.errors import PyroError,NamingError
group = ':test' # the namespace group for all test servers
######## main program
# objClass = the class of the Pyro implementation object
# objName = the name for the object to register in the Name Server
# delegate = whether to use the delegate approach or the regular
# objBase subclassing approach
def start(objClass, objName, delegate=0):
# initialize the server and set the default namespace group
Pyro.core.initServer()
Pyro.config.PYRO_NS_DEFAULTGROUP=group
# locate the NS
daemon = Pyro.core.Daemon()
locator = Pyro.naming.NameServerLocator()
print 'searching for Naming Service...'
ns = locator.getNS()
print 'Naming Service found at',ns.URI.address,'('+(Pyro.protocol.getHostname(ns.URI.address) or '??')+') port',ns.URI.port
# make sure our namespace group exists
try:
ns.createGroup(group)
except NamingError:
pass
daemon.useNameServer(ns)
# connect a new object implementation (first unregister previous one)
try:
ns.unregister(objName)
except NamingError:
pass
if delegate:
print 'Delegation...'
# use Deletation approach
obj=Pyro.core.ObjBase()
obj.delegateTo(objClass())
daemon.connect(obj,objName)
else:
# use regular ObjBase subclassing approach
obj=objClass()
daemon.connect(obj,objName)
# enter the service loop.
print 'Server object "'+objName+'" ready.'
try:
# daemon.setTimeout(5)
daemon.requestLoop()
except KeyboardInterrupt:
print 'shutting down gracefully.'
daemon.disconnect(obj)
daemon.shutdown()
print 'Exiting.'
|