Source code for distlock2.test_node

Module **sesf.distlock2.test_node** has class TestNode that implements
an RPC wrapper for a node of an implementation of distlock2 service.
Run ``python 2 1 --use imp_0.node 2 1`` in a shell 
to start a TestNode object wrapping an implementation imp_0.node.Node.

import Pyro4
from sesf.util.pyro import (start_pyrodaemon)
from threading import Thread
import argparse, importlib, time

[docs]@Pyro4.expose class TestNode(): def __init__(self, myid, use): self.myid = myid self.use = use self.pyrodaemon = None self.node = None self.pyrodaemon_thread = Thread( target=start_pyrodaemon, args=(self, "sesf.distlock2.testnode_"+str(myid))) self.pyrodaemon_thread.start() def start_testnode(self): tp = Pyro4.Proxy("PYRONAME:sesf.distlock2.servicetester") print("got testerproxy", tp, flush=True) imp_nodemodule = importlib.import_module(self.use[0]) print("creating_node_from_module", imp_nodemodule) self.node = imp_nodemodule.Node(self.use, tp) return def acq(self): # import pdb; pdb.set_trace() self.node.acq() return def rel(self): self.node.rel() return def end(self): self.node.end() return @Pyro4.oneway def end_pyrodaemon(self): self.pyrodaemon.shutdown()
##### END class TestNode if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument("n", type=int, help="number of nodes, must be 2") parser.add_argument("myid", type=int, help="id of this user system; 0 or 1") parser.add_argument("--use", nargs=argparse.REMAINDER, help="one or more --use node_descriptors") args = parser.parse_args() testnode = TestNode(args.myid, args.use) testnode.pyrodaemon_thread.join() time.sleep(10)