1 """Management for daemon processes
3 This module provides some client support for the daemon management
4 provided in the pdm.srv module.
7 import socket, pickle, struct, select, threading
9 __all__ = ["client", "replclient"]
11 class protoerr(Exception):
15 if isinstance(spec, socket.socket):
20 sk = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
23 sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
24 sk.connect(("localhost", int(spec)))
26 sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
28 sk.connect((spec[:p], int(spec[p + 1:])))
30 raise Exception("Unknown target specification %r" % spec)
34 if sk is not None: sk.close()
38 def __init__(self, sk, proto = None):
41 line = self.readline()
43 raise protoerr("Illegal protocol signature")
51 return self.sk.fileno()
55 p = self.buf.find("\n")
58 self.buf = self.buf[p + 1:]
60 ret = self.sk.recv(1024)
65 def select(self, proto):
67 raise Exception("Illegal protocol specified: %r" % proto)
68 self.sk.send(proto + "\n")
70 if len(rep) < 1 or rep[0] != "+":
71 raise protoerr("Error reply when selecting protocol %s: %s" % (proto, rep[1:]))
76 def __exit__(self, *excinfo):
80 class replclient(client):
81 def __init__(self, sk):
82 super(replclient, self).__init__(sk, "repl")
86 ncode = code.replace("\n\n", "\n")
87 if ncode == code: break
89 while len(code) > 0 and code[-1] == "\n":
91 self.sk.send(code + "\n\n")
100 raise protoerr("Error reply: %s" % ln[1:])
102 raise protoerr("Illegal reply: %s" % ln)
104 class perfproxy(object):
105 def __init__(self, cl, id, proto):
109 self.subscribers = set()
111 def lookup(self, name):
112 self.cl.lock.acquire()
117 self.cl.lock.release()
118 (proto,) = self.cl.run("lookup", id, self.id, name)
119 proxy = perfproxy(self.cl, id, proto)
120 self.cl.proxies[id] = proxy
124 return self.cl.run("ls", self.id)[0]
127 return self.cl.run("readattr", self.id)[0]
130 return self.cl.run("attrinfo", self.id)[0]
132 def invoke(self, method, *args, **kwargs):
133 return self.cl.run("invoke", self.id, method, args, kwargs)[0]
135 def subscribe(self, cb):
136 if cb in self.subscribers:
137 raise ValueError("Already subscribed")
138 if len(self.subscribers) == 0:
139 self.cl.run("subs", self.id)
140 self.subscribers.add(cb)
142 def unsubscribe(self, cb):
143 if cb not in self.subscribers:
144 raise ValueError("Not subscribed")
145 self.subscribers.remove(cb)
146 if len(self.subscribers) == 0:
147 self.cl.run("unsubs", self.id)
149 def notify(self, ev):
150 for cb in self.subscribers:
156 if self.id is not None:
157 self.cl.run("unbind", self.id)
158 del self.cl.proxies[self.id]
167 def __exit__(self, *excinfo):
171 class perfclient(client):
172 def __init__(self, sk):
173 super(perfclient, self).__init__(sk, "perf")
175 self.lock = threading.Lock()
180 buf = pickle.dumps(ob)
181 buf = struct.pack(">l", len(buf)) + buf
184 def recvb(self, num):
186 while len(buf) < num:
187 data = self.sk.recv(num - len(buf))
194 return pickle.loads(self.recvb(struct.unpack(">l", self.recvb(4))[0]))
196 def event(self, id, ev):
197 proxy = self.proxies.get(id)
198 if proxy is None: return
201 def dispatch(self, timeout = None):
202 rfd, wfd, efd = select.select([self.sk], [], [], timeout)
206 self.event(msg[1], msg[2])
208 raise ValueError("Unexpected non-event message: %r" % msg[0])
213 if reply[0] in ("+", "-"):
215 elif reply[0] == "*":
216 self.event(reply[1], reply[2])
218 raise ValueError("Illegal reply header: %r" % reply[0])
220 def run(self, cmd, *args):
223 self.send((cmd,) + args)
224 reply = self.recvreply()
232 def lookup(self, module, obnm):
239 (proto,) = self.run("bind", id, module, obnm)
240 proxy = perfproxy(self, id, proto)
241 self.proxies[id] = proxy
244 def find(self, name):
245 ret = self.names.get(name)
249 ret = self.find(name[:p]).lookup(name[p + 1:])
252 ret = self.lookup(name[:p], name[p + 1:])
253 self.names[name] = ret