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")
52 p = self.buf.find("\n")
55 self.buf = self.buf[p + 1:]
57 ret = self.sk.recv(1024)
62 def select(self, proto):
64 raise Exception("Illegal protocol specified: %r" % proto)
65 self.sk.send(proto + "\n")
67 if len(rep) < 1 or rep[0] != "+":
68 raise protoerr("Error reply when selecting protocol %s: %s" % (proto, rep[1:]))
73 def __exit__(self, *excinfo):
77 class replclient(client):
78 def __init__(self, sk):
79 super(replclient, self).__init__(sk, "repl")
83 ncode = code.replace("\n\n", "\n")
84 if ncode == code: break
86 while len(code) > 0 and code[-1] == "\n":
88 self.sk.send(code + "\n\n")
97 raise protoerr("Error reply: %s" % ln[1:])
99 raise protoerr("Illegal reply: %s" % ln)
101 class perfproxy(object):
102 def __init__(self, cl, id, proto):
106 self.subscribers = set()
108 def lookup(self, name):
109 self.cl.lock.acquire()
114 self.cl.lock.release()
115 (proto,) = self.cl.run("lookup", id, self.id, name)
116 proxy = perfproxy(self.cl, id, proto)
117 self.cl.proxies[id] = proxy
121 return self.cl.run("ls", self.id)[0]
124 return self.cl.run("readattr", self.id)[0]
127 return self.cl.run("attrinfo", self.id)[0]
129 def invoke(self, method, *args, **kwargs):
130 return self.cl.run("invoke", self.id, method, args, kwargs)[0]
132 def subscribe(self, cb):
133 if cb in self.subscribers:
134 raise ValueError("Already subscribed")
135 if len(self.subscribers) == 0:
136 self.cl.run("subs", self.id)
137 self.subscribers.add(cb)
139 def unsubscribe(self, cb):
140 if cb not in self.subscribers:
141 raise ValueError("Not subscribed")
142 self.subscribers.remove(cb)
143 if len(self.subscribers) == 0:
144 self.cl.run("unsubs", self.id)
146 def notify(self, ev):
147 for cb in self.subscribers:
153 if self.id is not None:
154 self.cl.run("unbind", self.id)
155 del self.cl.proxies[self.id]
164 def __exit__(self, *excinfo):
168 class perfclient(client):
169 def __init__(self, sk):
170 super(perfclient, self).__init__(sk, "perf")
172 self.lock = threading.Lock()
177 buf = pickle.dumps(ob)
178 buf = struct.pack(">l", len(buf)) + buf
181 def recvb(self, num):
183 while len(buf) < num:
184 data = self.sk.recv(num - len(buf))
191 return pickle.loads(self.recvb(struct.unpack(">l", self.recvb(4))[0]))
193 def event(self, id, ev):
194 proxy = self.proxies.get(id)
195 if proxy is None: return
198 def dispatch(self, timeout = None):
199 rfd, wfd, efd = select.select([self.sk], [], [], timeout)
203 self.event(msg[1], msg[2])
205 raise ValueError("Unexpected non-event message: %r" % msg[0])
210 if reply[0] in ("+", "-"):
212 elif reply[0] == "*":
213 self.event(reply[1], reply[2])
215 raise ValueError("Illegal reply header: %r" % reply[0])
217 def run(self, cmd, *args):
220 self.send((cmd,) + args)
221 reply = self.recvreply()
229 def lookup(self, module, obnm):
236 (proto,) = self.run("bind", id, module, obnm)
237 proxy = perfproxy(self, id, proto)
238 self.proxies[id] = proxy
241 def find(self, name):
242 ret = self.names.get(name)
246 ret = self.find(name[:p]).lookup(name[p + 1:])
249 ret = self.lookup(name[:p], name[p + 1:])
250 self.names[name] = ret